Filter AWS EventBridge Events for Genesys Cloud Conversation End by Queue

What You Will Build

• You will configure an AWS EventBridge Rule that filters Genesys Cloud conversation.end events to trigger a Lambda function only for a specific queue.
• You will use the Genesys Cloud EventBridge integration API to verify the target ARN and the AWS EventBridge Console API to define the input pattern.
• The tutorial covers Python (Boto3) for infrastructure verification and JSON for EventBridge input pattern definition.

Prerequisites

• Genesys Cloud: An active organization with EventBridge integration enabled. You need an OAuth client with analytics:events:read and integration:events:read scopes.
• AWS Account: Permissions to create EventBridge Rules and Lambda functions. You need the ARN of your target Lambda function.
• Python 3.9+: With boto3 and requests installed.
• Queue ID: The internal Genesys Cloud ID of the queue you want to filter (e.g., a1b2c3d4-e5f6-7890-abcd-ef1234567890).

Authentication Setup

You need two distinct authentication contexts: Genesys Cloud OAuth for API calls and AWS IAM credentials for Boto3.

Genesys Cloud OAuth

Use a standard Client Credentials flow. Store the access token in an environment variable for reuse.

import requests
import json
import os

def get_genesys_token(client_id: str, client_secret: str) -> str:
    """
    Retrieves a Genesys Cloud OAuth access token.
    """
    url = "https://api.mypurecloud.com/oauth/token"
    data = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret
    }
    
    response = requests.post(url, data=data)
    
    if response.status_code != 200:
        raise Exception(f"Failed to get token: {response.status_code} - {response.text}")
    
    token_data = response.json()
    return token_data["access_token"]

# Usage
GENESYS_TOKEN = get_genesys_token(
    os.getenv("GENESYS_CLIENT_ID"), 
    os.getenv("GENESYS_CLIENT_SECRET")
)

AWS Credentials

Configure your AWS credentials using standard AWS CLI configuration or environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_DEFAULT_REGION). Boto3 will automatically pick these up.

Implementation

Step 1: Identify the EventBridge Source and Target ARN

Before filtering, you must confirm the exact Source ARN or Event Source identifier Genesys Cloud uses. This ensures you do not accidentally filter out other critical events or create a rule that never matches.

We query the Genesys Cloud EventBridge integration to find the target ARN associated with your organization.

import requests

def get_eventbridge_targets(access_token: str) -> list:
    """
    Retrieves the list of EventBridge targets configured in Genesys Cloud.
    """
    url = "https://api.mypurecloud.com/api/v2/integrations/eventbridge/targets"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }
    
    response = requests.get(url, headers=headers)
    
    if response.status_code != 200:
        raise Exception(f"Failed to fetch targets: {response.status_code} - {response.text}")
    
    return response.json()["entities"]

targets = get_eventbridge_targets(GENESYS_TOKEN)
for target in targets:
    print(f"Target ARN: {target['arn']}")
    print(f"Target Status: {target['state']}")

Required Scope: integration:events:read

Expected Response:

[
  {
    "id": "abc123",
    "arn": "arn:aws:lambda:us-east-1:123456789012:function:MyGenesysProcessor",
    "state": "ACTIVE",
    "eventTypes": ["conversation.start", "conversation.end", "interaction.start"]
  }
]

Step 2: Define the EventBridge Input Pattern

This is the core logic. EventBridge filters events based on a JSON path match. Genesys Cloud sends events to EventBridge with a specific structure.

The conversation.end event payload from Genesys Cloud looks like this in EventBridge:

{
  "source": "com.genesyscloud.platform",
  "detail-type": "conversation.end",
  "detail": {
    "conversationId": "uuid-here",
    "queueId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    "routingType": "queue",
    "mediaType": "voice",
    "direction": "inbound"
  }
}

To filter for a specific queue, you must filter on detail.queueId.

Create a file named eventbridge_filter.json:

{
  "source": ["com.genesyscloud.platform"],
  "detail-type": ["conversation.end"],
  "detail": {
    "queueId": ["YOUR_SPECIFIC_QUEUE_ID_HERE"]
  }
}

Critical Notes:

1. Array Values: EventBridge input patterns require arrays for values, even if you only have one match. Use ["queue-id"], not "queue-id".
2. Field Existence: If the conversation was not routed through a queue (e.g., a direct transfer), queueId may be missing. The filter will correctly exclude these events.
3. Case Sensitivity: Field names are case-sensitive. queueId is correct; queueid will fail.

Step 3: Create the EventBridge Rule via Boto3

Now, apply this filter using AWS Boto3. This step creates the rule that connects Genesys Cloud events to your Lambda, but only for the specified queue.

import boto3
import json
import os

def create_filtered_eventbridge_rule(queue_id: str, lambda_arn: str, rule_name: str) -> dict:
    """
    Creates an EventBridge rule that filters Genesys conversation.end events for a specific queue.
    
    Args:
        queue_id: The Genesys Cloud Queue ID to filter on.
        lambda_arn: The ARN of the target Lambda function.
        rule_name: The name for the new EventBridge rule.
        
    Returns:
        The response from the put_rule API call.
    """
    events_client = boto3.client('events')
    
    # Define the input pattern
    input_pattern = {
        "source": ["com.genesyscloud.platform"],
        "detail-type": ["conversation.end"],
        "detail": {
            "queueId": [queue_id]
        }
    }
    
    try:
        # 1. Create the Rule
        response = events_client.put_rule(
            Name=rule_name,
            EventPattern=json.dumps(input_pattern),
            State='ENABLED',
            Description=f'Filter Genesys conversation.end for queue {queue_id}'
        )
        
        print(f"Rule created: {response['RuleArn']}")
        
        # 2. Add the Target (Lambda)
        # Note: You need to ensure the Lambda has an EventBridge permission policy
        # This is often handled by the 'aws_lambda_permission' resource in CloudFormation
        # or via 'add_permission' API call. Here we just attach the target.
        
        events_client.put_targets(
            Rule=rule_name,
            Targets=[
                {
                    'Id': '1',
                    'Arn': lambda_arn
                }
            ]
        )
        
        print(f"Target added: {lambda_arn}")
        return response
        
    except events_client.exceptions.ResourceNotFoundException:
        raise Exception("EventBridge service not available in this region.")
    except Exception as e:
        raise Exception(f"Failed to create rule: {str(e)}")

# Usage
QUEUE_ID = "a1b2c3d4-e5f6-7890-abcd-ef1234567890" # Replace with your Queue ID
LAMBDA_ARN = "arn:aws:lambda:us-east-1:123456789012:function:MyGenesysProcessor" # Replace with your Lambda ARN
RULE_NAME = "GenesysConversationEndSpecificQueue"

# Ensure Lambda has permission to be invoked by EventBridge
lambda_client = boto3.client('lambda')
try:
    lambda_client.add_permission(
        FunctionName=LAMBDA_ARN.split(':')[-1], # Extract function name from ARN
        StatementId='AllowEventBridgeInvoke',
        Action='lambda:InvokeFunction',
        Principal='events.amazonaws.com',
        SourceArn=f"arn:aws:events:{os.getenv('AWS_DEFAULT_REGION')}:123456789012:rule/{RULE_NAME}"
    )
    print("Lambda permission added.")
except lambda_client.exceptions.ResourceConflictException:
    print("Lambda permission already exists.")

create_filtered_eventbridge_rule(QUEUE_ID, LAMBDA_ARN, RULE_NAME)

Required AWS Permissions:

• events:PutRule
• events:PutTargets
• lambda:AddPermission

Step 4: Verify the Filter in AWS Console

To ensure the filter is applied correctly, you can list the rules and inspect the event pattern.

def describe_eventbridge_rule(rule_name: str) -> dict:
    """
    Retrieves details of an existing EventBridge rule.
    """
    events_client = boto3.client('events')
    
    try:
        response = events_client.describe_rule(Name=rule_name)
        print(f"Rule Name: {response['Name']}")
        print(f"State: {response['State']}")
        print(f"Event Pattern: {response['EventPattern']}")
        return response
    except events_client.exceptions.ResourceNotFoundException:
        print(f"Rule {rule_name} not found.")
        return None

describe_eventbridge_rule(RULE_NAME)

Complete Working Example

This script combines authentication, target verification, and rule creation into a single executable module.

import requests
import boto3
import json
import os
import sys

def get_genesys_token(client_id: str, client_secret: str) -> str:
    url = "https://api.mypurecloud.com/oauth/token"
    data = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret
    }
    response = requests.post(url, data=data)
    if response.status_code != 200:
        raise Exception(f"Token error: {response.text}")
    return response.json()["access_token"]

def verify_genesys_target(access_token: str) -> str:
    """Returns the first active EventBridge target ARN."""
    url = "https://api.mypurecloud.com/api/v2/integrations/eventbridge/targets"
    headers = {"Authorization": f"Bearer {access_token}"}
    response = requests.get(url, headers=headers)
    if response.status_code != 200:
        raise Exception(f"Target fetch error: {response.text}")
    
    targets = response.json().get("entities", [])
    for t in targets:
        if t.get("state") == "ACTIVE":
            return t["arn"]
    raise Exception("No active EventBridge targets found in Genesys Cloud.")

def setup_eventbridge_filter(queue_id: str, lambda_arn: str, rule_name: str):
    events_client = boto3.client('events')
    lambda_client = boto3.client('lambda')
    region = os.getenv('AWS_DEFAULT_REGION', 'us-east-1')
    
    # 1. Define Filter
    input_pattern = {
        "source": ["com.genesyscloud.platform"],
        "detail-type": ["conversation.end"],
        "detail": {
            "queueId": [queue_id]
        }
    }
    
    # 2. Create Rule
    try:
        events_client.put_rule(
            Name=rule_name,
            EventPattern=json.dumps(input_pattern),
            State='ENABLED',
            Description=f'Genesys conversation.end for queue {queue_id}'
        )
        print(f"[Success] Rule '{rule_name}' created.")
    except Exception as e:
        print(f"[Error] Failed to create rule: {e}")
        return

    # 3. Add Target
    try:
        events_client.put_targets(
            Rule=rule_name,
            Targets=[{'Id': '1', 'Arn': lambda_arn}]
        )
        print(f"[Success] Target added to rule.")
    except Exception as e:
        print(f"[Error] Failed to add target: {e}")
        return

    # 4. Add Lambda Permission
    func_name = lambda_arn.split(':')[-1]
    rule_arn = f"arn:aws:events:{region}:{os.getenv('AWS_ACCOUNT_ID')}:rule/{rule_name}"
    
    try:
        lambda_client.add_permission(
            FunctionName=func_name,
            StatementId='AllowEventBridgeInvokeSpecificQueue',
            Action='lambda:InvokeFunction',
            Principal='events.amazonaws.com',
            SourceArn=rule_arn
        )
        print(f"[Success] Lambda permission added.")
    except lambda_client.exceptions.ResourceConflictException:
        print("[Info] Lambda permission already exists.")
    except Exception as e:
        print(f"[Error] Failed to add Lambda permission: {e}")

if __name__ == "__main__":
    # Configuration
    GENESYS_CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
    GENESYS_CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
    TARGET_QUEUE_ID = os.getenv("TARGET_QUEUE_ID") # e.g., a1b2c3d4-...
    TARGET_LAMBDA_ARN = os.getenv("TARGET_LAMBDA_ARN")
    RULE_NAME = os.getenv("RULE_NAME", "GenesysQueueFilter")

    if not all([GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, TARGET_QUEUE_ID, TARGET_LAMBDA_ARN]):
        print("Missing environment variables.")
        sys.exit(1)

    try:
        token = get_genesys_token(GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET)
        target_arn = verify_genesys_target(token)
        print(f"Verified Genesys Target: {target_arn}")
        
        setup_eventbridge_filter(TARGET_QUEUE_ID, TARGET_LAMBDA_ARN, RULE_NAME)
        
    except Exception as e:
        print(f"Fatal Error: {e}")
        sys.exit(1)

Common Errors & Debugging

Error: Rule Created but Lambda Not Invoked

Cause: The EventBridge rule is created, but the Lambda function lacks the permission to be invoked by EventBridge, or the queueId in the Genesys event does not match the filter exactly.

Fix:

1. Check CloudWatch Logs for the Lambda function. Look for User arn:aws:events:... is not authorized to invoke function.
2. If this error appears, run the add_permission step again. Ensure the SourceArn matches the exact ARN of the EventBridge rule.
3. Verify the queueId in the filter. Genesys Cloud Queue IDs are UUIDs. If you copied the ID from the URL, ensure it is the full UUID, not a truncated version.

Error: ResourceNotFoundException for EventBridge

Cause: The AWS region specified in your AWS credentials does not support EventBridge or the rule name already exists in a different state.

Fix:

1. Ensure AWS_DEFAULT_REGION is set to a region where EventBridge is available (most major regions).
2. Check if the rule already exists using describe_eventbridge_rule. If it does, update it instead of creating a new one.

Error: queueId Not Present in Event

Cause: The conversation ended without being associated with a queue (e.g., a direct transfer, a chat transfer, or an internal note).

Fix:

1. If you need to capture all conversation ends regardless of queue, remove the queueId filter from the input pattern.
2. If you strictly need queue-based filtering, this is expected behavior. The event is correctly filtered out.

Error: 401 Unauthorized on Genesys API

Cause: The OAuth token is expired or the client credentials are incorrect.

Fix:

1. Verify GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET in your environment variables.
2. Ensure the OAuth client has the integration:events:read scope.

Official References

• Genesys Cloud EventBridge Integration API
• AWS EventBridge Input Pattern Syntax
• Genesys Cloud Conversation Event Schema
• Boto3 Events Client Documentation