Filter EventBridge Events for Conversation End by Queue in Genesys Cloud

What You Will Build

• You will configure a Genesys Cloud event subscription that pushes conversation.end events to an AWS EventBridge bus.
• You will implement an AWS EventBridge Rule with a precise event pattern to filter these events so that only conversations ending in a specific Genesys Cloud queue trigger a downstream Lambda function.
• This tutorial covers the Python SDK for Genesys Cloud configuration and the AWS CLI/Boto3 for EventBridge rule creation.

Prerequisites

• Genesys Cloud OAuth Client: A confidential client (Client Credentials flow) with the following scopes:
  • event:subscribe:write (to create the event subscription)
  • queue:read (to retrieve queue IDs)
  • conversation:read (optional, for debugging context)
• AWS Account: An account with an existing EventBridge bus (default or custom) and permissions to create EventBridge rules and invoke Lambda functions.
• Python Runtime: Python 3.9+ with pip.
• Dependencies:

  pip install purecloudplatformclientv2 boto3
  

• Environment Variables: Set the following in your shell or .env file:
  • GENESYS_CLOUD_CLIENT_ID
  • GENESYS_CLOUD_CLIENT_SECRET
  • GENESYS_CLOUD_REGION (e.g., us-east-1)
  • AWS_ACCESS_KEY_ID
  • AWS_SECRET_ACCESS_KEY
  • AWS_DEFAULT_REGION

Authentication Setup

Genesys Cloud uses OAuth 2.0 Client Credentials flow for server-to-server integrations. You must obtain an access token before making API calls.

Python Authentication Helper

This function retrieves a token and handles basic error logging. In production, you should cache this token and refresh it before expiration (typically 1 hour).

import os
import requests
from purecloudplatformclientv2 import PlatformClient

def get_genesys_platform_client() -> PlatformClient:
    """
    Initializes and authenticates the Genesys Cloud PlatformClient.
    """
    client_id = os.environ.get('GENESYS_CLOUD_CLIENT_ID')
    client_secret = os.environ.get('GENESYS_CLOUD_CLIENT_SECRET')
    region = os.environ.get('GENESYS_CLOUD_REGION', 'us-east-1')

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET must be set.")

    platform_client = PlatformClient()
    
    # Configure OAuth2
    oauth_config = platform_client.oauth2_config
    oauth_config.client_id = client_id
    oauth_config.client_secret = client_secret
    
    # Set the region (e.g., 'us-east-1' maps to 'mypurecloud.com')
    # Note: The SDK handles region mapping, but explicit setting is clearer for debugging
    platform_client.set_base_url(f'https://{region}.mypurecloud.com')

    try:
        # This triggers the token fetch
        platform_client.oauth2_config.get_access_token()
        return platform_client
    except Exception as e:
        raise RuntimeError(f"Failed to authenticate with Genesys Cloud: {e}")

Implementation

Step 1: Identify the Target Queue ID

EventBridge filters rely on exact string matching for IDs. You must know the UUID of the queue you want to filter on.

def get_queue_by_name(platform_client: PlatformClient, queue_name: str) -> str:
    """
    Retrieves the Queue ID by name.
    Requires scope: queue:read
    """
    queues_api = platform_client.QueuesApi()
    
    try:
        # List queues. In large orgs, consider paging or searching by specific attributes
        response = queues_api.get_queues()
        
        for queue in response.entities:
            if queue.name.lower() == queue_name.lower():
                print(f"Found Queue: {queue.name} (ID: {queue.id})")
                return queue.id
        
        raise ValueError(f"Queue '{queue_name}' not found.")
        
    except Exception as e:
        raise RuntimeError(f"Error fetching queues: {e}")

Step 2: Configure the Genesys Cloud Event Subscription

You must create an event subscription in Genesys Cloud that sends conversation.end events to your AWS EventBridge endpoint. Genesys Cloud supports AWS EventBridge as a direct destination.

Critical Detail: Genesys Cloud sends a generic conversation.end event. It does not allow filtering by queue at the source (Genesys side) for EventBridge destinations in the same granular way it does for webhooks with query parameters. Therefore, you must subscribe to all conversation.end events and filter them in AWS EventBridge.

from purecloudplatformclientv2 import EventSubscriptionRequest, AwsEventBridgeDestination

def create_event_subscription(platform_client: PlatformClient, queue_id: str) -> str:
    """
    Creates an EventBridge subscription for conversation.end events.
    Requires scope: event:subscribe:write
    """
    events_api = platform_client.EventsApi()
    
    # Define the destination
    # Replace with your actual AWS EventBridge Bus Name and ARN if using a custom bus
    # For standard integration, Genesys handles the ARN resolution if you provide the account ID and region
    aws_destination = AwsEventBridgeDestination()
    aws_destination.account_id = os.environ.get('AWS_ACCOUNT_ID') # Add this env var
    aws_destination.region = os.environ.get('AWS_DEFAULT_REGION')
    aws_destination.bus_name = "default" # Or your custom bus name
    
    # Define the subscription
    sub_request = EventSubscriptionRequest()
    sub_request.name = "ConversationEnd-EventBridge-Integration"
    sub_request.description = "Sends all conversation.end events to AWS EventBridge"
    sub_request.enabled = True
    
    # Set the destination
    sub_request.destination = aws_destination
    
    # Define the event type
    # We subscribe to conversation.end. Note: We cannot filter by queue here.
    sub_request.event_types = ["conversation.end"]
    
    # Optional: Add metadata mapping if needed, but usually default is sufficient
    # sub_request.metadata_mappings = [...]

    try:
        response = events_api.post_events_subscriptions(body=sub_request)
        print(f"Subscription created successfully. ID: {response.id}")
        return response.id
    except Exception as e:
        raise RuntimeError(f"Failed to create event subscription: {e}")

Step 3: Create the AWS EventBridge Rule with Filter Pattern

This is the core of the solution. EventBridge allows you to define an EventPattern that matches specific fields in the incoming event payload.

When Genesys Cloud sends a conversation.end event to EventBridge, the payload structure follows the AWS CloudWatch Event format. The Genesys-specific data is contained within the detail object.

The Payload Structure:

{
  "version": "0",
  "id": "genesys-event-id",
  "detail-type": "conversation.end",
  "source": "genesys",
  "account": "your-aws-account-id",
  "time": "2023-10-27T10:00:00Z",
  "region": "us-east-1",
  "resources": [],
  "detail": {
    "conversationId": "abc-123-def",
    "queueId": "uuid-of-your-target-queue", 
    "wrapUpCode": "Resolved",
    "duration": 120
    // ... other conversation attributes
  }
}

To filter for a specific queue, you must match the detail.queueId field.

Using AWS CLI (Recommended for Idempotency)

Create a file named eventbridge-rule.json:

{
  "Source": [
    "genesys"
  ],
  "Detail-Type": [
    "conversation.end"
  ],
  "Detail": {
    "queueId": [
      "<REPLACE_WITH_YOUR_QUEUE_ID>"
    ]
  }
}

Replace <REPLACE_WITH_YOUR_QUEUE_ID> with the UUID obtained in Step 1.

Run the following AWS CLI command to create the rule:

aws events put-rule \
  --name GenesysConversationEndSpecificQueue \
  --event-pattern file://eventbridge-rule.json \
  --state ENABLED \
  --description "Triggers only for conversation.end events from specific Genesys Queue"

Using Boto3 (Python SDK)

If you prefer to manage this infrastructure as code via Python:

import boto3
import json

def create_eventbridge_rule(queue_id: str, target_arn: str):
    """
    Creates an EventBridge rule that filters for conversation.end events 
    for a specific queue.
    
    Args:
        queue_id: The Genesys Cloud Queue UUID.
        target_arn: The ARN of the Lambda function to invoke.
    """
    events_client = boto3.client('events')
    
    # Define the event pattern
    event_pattern = {
        "Source": ["genesys"],
        "Detail-Type": ["conversation.end"],
        "Detail": {
            "queueId": [queue_id]
        }
    }
    
    rule_name = "GenesysConversationEndSpecificQueue"
    
    try:
        # Create the rule
        response = events_client.put_rule(
            Name=rule_name,
            EventPattern=json.dumps(event_pattern),
            State='ENABLED',
            Description='Filters conversation.end events for specific Genesys Queue'
        )
        print(f"Rule created with ARN: {response['RuleArn']}")
        
        # Add the Lambda target
        events_client.put_targets(
            Rule=rule_name,
            Targets=[
                {
                    'Id': '1',
                    'Arn': target_arn,
                    'InputPath': '$' # Passes the entire event to the Lambda
                }
            ]
        )
        print("Lambda target added successfully.")
        
    except events_client.exceptions.ResourceNotFoundException:
        print("Error: EventBridge service not found in this region.")
    except Exception as e:
        raise RuntimeError(f"Failed to create EventBridge rule: {e}")

Step 4: Verify the Filter Logic

It is critical to understand that EventBridge filters are exact matches for strings unless you use wildcards (which are limited in EventBridge).

1. Exact Match: The queueId in the event payload must exactly match the UUID in the filter.
2. Case Sensitivity: UUIDs are case-insensitive in Genesys, but EventBridge string matching is case-sensitive. Ensure the UUID from Genesys (Step 1) and the UUID in the filter (Step 3) use the same casing.
3. Null Values: If a conversation ends without being routed to a queue (e.g., a direct agent call), the queueId field might be null or missing. The filter queueId: ["uuid"] will not match null values. This is desirable for filtering.

Complete Working Example

This script combines authentication, queue lookup, and EventBridge rule creation.

import os
import sys
import json
import boto3
from purecloudplatformclientv2 import PlatformClient

def main():
    # 1. Configuration
    QUEUE_NAME = "Sales Support" # Change to your queue name
    LAMBDA_ARN = "arn:aws:lambda:us-east-1:123456789012:function:MyGenesysHandler" # Change to your Lambda ARN
    AWS_ACCOUNT_ID = os.environ.get('AWS_ACCOUNT_ID') # Required for Genesys Subscription

    if not AWS_ACCOUNT_ID:
        print("Error: AWS_ACCOUNT_ID environment variable is required.")
        sys.exit(1)

    # 2. Authenticate with Genesys
    try:
        print("Authenticating with Genesys Cloud...")
        platform_client = get_genesys_platform_client()
    except Exception as e:
        print(f"Authentication failed: {e}")
        sys.exit(1)

    # 3. Get Queue ID
    try:
        print(f"Looking up Queue: {QUEUE_NAME}")
        queue_id = get_queue_by_name(platform_client, QUEUE_NAME)
    except Exception as e:
        print(f"Queue lookup failed: {e}")
        sys.exit(1)

    # 4. Create Genesys Event Subscription
    try:
        print("Creating Genesys Event Subscription...")
        sub_id = create_event_subscription(platform_client, queue_id)
    except Exception as e:
        print(f"Subscription creation failed: {e}")
        sys.exit(1)

    # 5. Create AWS EventBridge Rule
    try:
        print(f"Creating EventBridge Rule for Queue ID: {queue_id}")
        create_eventbridge_rule(queue_id, LAMBDA_ARN)
    except Exception as e:
        print(f"EventBridge rule creation failed: {e}")
        sys.exit(1)

    print("Setup complete. EventBridge will now filter conversation.end events for the specified queue.")

# --- Helper Functions from Previous Sections ---

def get_genesys_platform_client() -> PlatformClient:
    client_id = os.environ.get('GENESYS_CLOUD_CLIENT_ID')
    client_secret = os.environ.get('GENESYS_CLOUD_CLIENT_SECRET')
    region = os.environ.get('GENESYS_CLOUD_REGION', 'us-east-1')

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET must be set.")

    platform_client = PlatformClient()
    oauth_config = platform_client.oauth2_config
    oauth_config.client_id = client_id
    oauth_config.client_secret = client_secret
    platform_client.set_base_url(f'https://{region}.mypurecloud.com')

    try:
        platform_client.oauth2_config.get_access_token()
        return platform_client
    except Exception as e:
        raise RuntimeError(f"Failed to authenticate with Genesys Cloud: {e}")

def get_queue_by_name(platform_client: PlatformClient, queue_name: str) -> str:
    queues_api = platform_client.QueuesApi()
    try:
        response = queues_api.get_queues()
        for queue in response.entities:
            if queue.name.lower() == queue_name.lower():
                return queue.id
        raise ValueError(f"Queue '{queue_name}' not found.")
    except Exception as e:
        raise RuntimeError(f"Error fetching queues: {e}")

def create_event_subscription(platform_client: PlatformClient, queue_id: str) -> str:
    events_api = platform_client.EventsApi()
    aws_destination = AwsEventBridgeDestination()
    aws_destination.account_id = os.environ.get('AWS_ACCOUNT_ID')
    aws_destination.region = os.environ.get('AWS_DEFAULT_REGION')
    aws_destination.bus_name = "default"

    sub_request = EventSubscriptionRequest()
    sub_request.name = "ConversationEnd-EventBridge-Integration"
    sub_request.enabled = True
    sub_request.destination = aws_destination
    sub_request.event_types = ["conversation.end"]

    try:
        response = events_api.post_events_subscriptions(body=sub_request)
        return response.id
    except Exception as e:
        raise RuntimeError(f"Failed to create event subscription: {e}")

def create_eventbridge_rule(queue_id: str, target_arn: str):
    events_client = boto3.client('events')
    event_pattern = {
        "Source": ["genesys"],
        "Detail-Type": ["conversation.end"],
        "Detail": {
            "queueId": [queue_id]
        }
    }
    rule_name = "GenesysConversationEndSpecificQueue"
    
    try:
        events_client.put_rule(
            Name=rule_name,
            EventPattern=json.dumps(event_pattern),
            State='ENABLED'
        )
        events_client.put_targets(
            Rule=rule_name,
            Targets=[{'Id': '1', 'Arn': target_arn, 'InputPath': '$'}]
        )
    except Exception as e:
        raise RuntimeError(f"Failed to create EventBridge rule: {e}")

if __name__ == "__main__":
    # Import missing classes for the complete example
    from purecloudplatformclientv2 import AwsEventBridgeDestination, EventSubscriptionRequest
    main()

Common Errors & Debugging

Error: 403 Forbidden on Genesys API Calls

• Cause: The OAuth client lacks the required scopes.
• Fix: Ensure the client has event:subscribe:write and queue:read. If you are using a user-based token, ensure the user has the “Event Subscriptions” permission in the Genesys Admin UI.

Error: EventBridge Rule Not Triggering

• Cause 1: The queueId in the filter does not match the queueId in the event.
  • Fix: Use CloudWatch Logs Insights to query the EventBridge delivery logs. Compare the detail.queueId from a successful test event with the filter value. Ensure casing matches.
• Cause 2: The event type is incorrect.
  • Fix: Verify the Genesys subscription is for conversation.end. If you are testing with conversation.start, the rule will not trigger.
• Cause 3: The Lambda function is not in the same region as the EventBridge rule.
  • Fix: EventBridge rules must target resources in the same AWS region.

Error: InvalidEventPatternException

• Cause: The JSON structure of the event pattern is invalid.
• Fix: Ensure the Detail object keys match the actual field names in the Genesys payload. The field is queueId, not QueueId or queue_id.

Official References

• Genesys Cloud Event Subscriptions API Documentation
• AWS EventBridge Event Patterns Documentation
• Genesys Cloud AWS EventBridge Integration Guide