Implementing Event Fan-Out from Genesys Cloud EventBridge in Python

What You Will Build

A Python event consumer that ingests Genesys Cloud interaction completion events via AWS EventBridge, evaluates them against a configurable rule engine, routes matches to priority-based SQS queues, handles failures with a dead-letter queue, and instruments processing latency with OpenTelemetry spans. The code uses the AWS SDK for Python (boto3) for queue operations and the OpenTelemetry Python SDK for distributed tracing.

Prerequisites

• Genesys Cloud Integration Scope: event:write (required to configure the EventBridge integration via /api/v2/integrations/eventbridge)
• AWS IAM Permissions: sqs:ReceiveMessage, sqs:DeleteMessage, sqs:SendMessage, sqs:GetQueueAttributes, logs:CreateLogGroup
• Runtime: Python 3.10 or higher
• Dependencies: boto3, opentelemetry-api, opentelemetry-sdk, opentelemetry-exporter-otlp, opentelemetry-instrumentation-boto3, jsonpath-ng
• Architecture Note: Genesys Cloud EventBridge pushes events to AWS EventBridge. You must create an EventBridge rule filtering on source: genesys.cloud and detail-type: Interaction Completed, with an SQS queue as the target. This Python consumer polls that SQS queue.

Authentication Setup

The consumer relies on the AWS Security Token Service (STS) credential provider chain. Deploy the script on an EC2 instance with an attached IAM role, or set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY for local testing. Boto3 automatically handles credential rotation. The Genesys Cloud EventBridge integration itself requires an OAuth 2.0 client with the event:write scope to enable the bridge in the Genesys Admin Console.

import os
import boto3
from botocore.config import Config
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.boto3 import Boto3Instrumentor

# Initialize OpenTelemetry tracing
trace.set_tracer_provider(TracerProvider())
otel_endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint=otel_endpoint))
)
tracer = trace.get_tracer("genesys-eventbridge-consumer")

# Instrument boto3 for automatic AWS API tracing
Boto3Instrumentor().instrument()

# Configure AWS session with retry strategy for transient 5xx and 429 errors
aws_config = Config(
    retries={
        "max_attempts": 5,
        "mode": "adaptive"
    },
    read_timeout=10,
    connect_timeout=5
)

session = boto3.Session()
sqs_client = session.client("sqs", config=aws_config)

Implementation

Step 1: Event Ingestion and Batch Processing

EventBridge delivers events to SQS in batches. The consumer must poll the source queue, parse the Genesys Cloud payload structure, and preserve message handles for deletion. The Genesys Cloud interaction completion event follows the CloudEvents specification with a detail object containing interaction metadata.

from typing import Dict, Any, List
import logging

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

SOURCE_QUEUE_URL = os.getenv("GENESYS_SOURCE_QUEUE_URL")
DLQ_URL = os.getenv("GENESYS_DLQ_URL")
MAX_MESSAGES = 10

def fetch_events() -> List[Dict[str, Any]]:
    """Polls the EventBridge target SQS queue for Genesys interaction completion events."""
    with tracer.start_as_current_span("sqs.receive_batch") as span:
        span.set_attribute("queue.url", SOURCE_QUEUE_URL)
        span.set_attribute("max.messages", MAX_MESSAGES)
        
        try:
            response = sqs_client.receive_message(
                QueueUrl=SOURCE_QUEUE_URL,
                MaxNumberOfMessages=MAX_MESSAGES,
                WaitTimeSeconds=5
            )
            messages = response.get("Messages", [])
            span.set_attribute("messages.received", len(messages))
            return messages
        except Exception as e:
            span.record_exception(e)
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            logger.error("Failed to receive messages: %s", e)
            raise

Step 2: Rule Engine Evaluation and Priority Routing

The rule engine evaluates complex filter expressions against the event payload. We use jsonpath-ng for safe path extraction and a condition parser that supports equality, comparison, and set membership. Rules are defined as JSON objects mapping conditions to target queue URLs.

from jsonpath_ng import parse
from jsonpath_ng.exceptions import JsonPathParserError
import math

RULES = [
    {
        "id": "critical_priority",
        "condition": "detail.tags.priority == 'critical'",
        "target_queue": os.getenv("CRITICAL_QUEUE_URL")
    },
    {
        "id": "high_priority",
        "condition": "detail.tags.priority == 'high' or detail.duration_ms > 120000",
        "target_queue": os.getenv("HIGH_QUEUE_URL")
    },
    {
        "id": "standard_priority",
        "condition": "detail.channel == 'voice' and detail.outcome == 'completed'",
        "target_queue": os.getenv("STANDARD_QUEUE_URL")
    }
]

def evaluate_condition(condition_str: str, event_detail: Dict[str, Any]) -> bool:
    """Safely evaluates a filter expression against the event detail object."""
    try:
        # Replace JSON paths with extracted values for safe evaluation
        jsonpath_patterns = [
            ("detail.tags.priority", "priority"),
            ("detail.duration_ms", "duration_ms"),
            ("detail.channel", "channel"),
            ("detail.outcome", "outcome")
        ]
        
        safe_condition = condition_str
        local_vars = {}
        
        for path, var_name in jsonpath_patterns:
            try:
                expr = parse(path)
                matches = expr.find(event_detail)
                value = matches[0].value if matches else None
                local_vars[var_name] = value
                safe_condition = safe_condition.replace(path, var_name)
            except JsonPathParserError:
                continue
                
        # Safe evaluation with restricted builtins
        allowed_names = {
            "priority": local_vars.get("priority"),
            "duration_ms": local_vars.get("duration_ms"),
            "channel": local_vars.get("channel"),
            "outcome": local_vars.get("outcome"),
            "True": True,
            "False": False,
            "None": None
        }
        
        return eval(safe_condition, {"__builtins__": {}}, allowed_names)
    except Exception as e:
        logger.warning("Rule evaluation failed for condition '%s': %s", condition_str, e)
        return False

def match_rules(event_detail: Dict[str, Any]) -> List[Dict[str, str]]:
    """Evaluates all rules and returns matching target queues."""
    matched = []
    for rule in RULES:
        if evaluate_condition(rule["condition"], event_detail):
            matched.append({"rule_id": rule["id"], "target_queue": rule["target_queue"]})
    return matched

Step 3: SQS Fan-Out, Dead-Letter Queue Handling, and Latency Tracking

Matched events are dispatched to multiple downstream queues. The implementation implements exponential backoff for throttling errors, records processing latency in OpenTelemetry spans, and routes unprocessable messages to a dead-letter queue.

import time
import uuid
from botocore.exceptions import ClientError

def send_to_queue(queue_url: str, message_body: str, span: trace.Span) -> bool:
    """Sends a message to a target SQS queue with retry logic for 429 and 5xx errors."""
    max_retries = 4
    base_delay = 0.5
    
    for attempt in range(max_retries + 1):
        try:
            sqs_client.send_message(
                QueueUrl=queue_url,
                MessageBody=message_body,
                MessageGroupId="interaction-events",  # FIFO queue support if needed
                MessageDeduplicationId=str(uuid.uuid4())
            )
            span.set_attribute("sqs.send.success", True)
            return True
        except ClientError as e:
            error_code = e.response["Error"]["Code"]
            retryable = error_code in ("ThrottlingException", "ServiceUnavailable", "InternalError")
            
            if not retryable:
                span.set_attribute("sqs.send.error", error_code)
                span.record_exception(e)
                return False
                
            if attempt < max_retries:
                delay = base_delay * (2 ** attempt) + (hash(queue_url) % 100) / 1000
                time.sleep(delay)
                span.add_event("retry.attempt", {"attempt": attempt + 1, "delay": delay})
            else:
                span.set_attribute("sqs.send.exhausted_retries", True)
                return False
        except Exception as e:
            span.record_exception(e)
            return False

def route_to_dlq(message: Dict[str, Any], error_reason: str) -> None:
    """Routes failed messages to the dead-letter queue."""
    try:
        sqs_client.send_message(
            QueueUrl=DLQ_URL,
            MessageBody=json.dumps({
                "original_message": message.get("Body"),
                "receipt_handle": message.get("ReceiptHandle"),
                "error": error_reason,
                "timestamp": time.time()
            })
        )
        logger.info("Routed message to DLQ: %s", error_reason)
    except Exception as e:
        logger.error("DLQ send failed: %s", e)

def process_event(message: Dict[str, Any]) -> None:
    """Processes a single Genesys Cloud event with fan-out and error handling."""
    with tracer.start_as_current_span("event.process") as span:
        span.set_attribute("message.id", message.get("MessageId"))
        span.set_attribute("event.source", "genesys.cloud")
        
        try:
            body = json.loads(message.get("Body", "{}"))
            detail = body.get("detail", {})
            
            if not detail:
                raise ValueError("Missing 'detail' field in Genesys event payload")
                
            span.set_attribute("interaction.id", detail.get("id"))
            span.set_attribute("interaction.channel", detail.get("channel"))
            
            matched_queues = match_rules(detail)
            
            if not matched_queues:
                span.set_attribute("rules.matched", 0)
                logger.info("No rules matched for interaction %s", detail.get("id"))
                # Delete from source queue even if no rules match
                sqs_client.delete_message(QueueUrl=SOURCE_QUEUE_URL, ReceiptHandle=message["ReceiptHandle"])
                return
                
            span.set_attribute("rules.matched", len(matched_queues))
            
            for target in matched_queues:
                queue_url = target["target_queue"]
                rule_id = target["rule_id"]
                
                with tracer.start_as_current_span("sqs.fanout", attributes={"rule.id": rule_id}) as fanout_span:
                    success = send_to_queue(queue_url, message["Body"], fanout_span)
                    if not success:
                        fanout_span.set_status(trace.Status(trace.StatusCode.ERROR, "Fan-out failed"))
                        raise RuntimeError(f"Failed to send to {queue_url}")
                        
            # Delete from source queue after successful fan-out
            sqs_client.delete_message(QueueUrl=SOURCE_QUEUE_URL, ReceiptHandle=message["ReceiptHandle"])
            span.set_attribute("processing.status", "success")
            
        except Exception as e:
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            span.record_exception(e)
            route_to_dlq(message, str(e))

Complete Working Example

The following script combines all components into a production-ready event consumer. It runs continuously, polls the source queue, processes batches, and handles lifecycle errors.

import json
import time
import os
import logging
import boto3
from botocore.config import Config
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.boto3 import Boto3Instrumentor
from jsonpath_ng import parse
from jsonpath_ng.exceptions import JsonPathParserError
from botocore.exceptions import ClientError
from typing import Dict, Any, List

# Logging setup
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# OpenTelemetry initialization
trace.set_tracer_provider(TracerProvider())
otel_endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint=otel_endpoint))
)
tracer = trace.get_tracer("genesys-eventbridge-consumer")
Boto3Instrumentor().instrument()

# AWS configuration
aws_config = Config(retries={"max_attempts": 5, "mode": "adaptive"}, read_timeout=10, connect_timeout=5)
session = boto3.Session()
sqs_client = session.client("sqs", config=aws_config)

# Queue URLs
SOURCE_QUEUE_URL = os.getenv("GENESYS_SOURCE_QUEUE_URL")
DLQ_URL = os.getenv("GENESYS_DLQ_URL")

# Rule definitions
RULES = [
    {"id": "critical_priority", "condition": "detail.tags.priority == 'critical'", "target_queue": os.getenv("CRITICAL_QUEUE_URL")},
    {"id": "high_priority", "condition": "detail.tags.priority == 'high' or detail.duration_ms > 120000", "target_queue": os.getenv("HIGH_QUEUE_URL")},
    {"id": "standard_priority", "condition": "detail.channel == 'voice' and detail.outcome == 'completed'", "target_queue": os.getenv("STANDARD_QUEUE_URL")}
]

def evaluate_condition(condition_str: str, event_detail: Dict[str, Any]) -> bool:
    try:
        safe_condition = condition_str
        local_vars = {}
        for path, var_name in [("detail.tags.priority", "priority"), ("detail.duration_ms", "duration_ms"), ("detail.channel", "channel"), ("detail.outcome", "outcome")]:
            try:
                expr = parse(path)
                matches = expr.find(event_detail)
                local_vars[var_name] = matches[0].value if matches else None
                safe_condition = safe_condition.replace(path, var_name)
            except JsonPathParserError:
                continue
        allowed_names = {k: local_vars.get(k) for k in local_vars}
        allowed_names.update({"True": True, "False": False, "None": None})
        return eval(safe_condition, {"__builtins__": {}}, allowed_names)
    except Exception:
        return False

def match_rules(event_detail: Dict[str, Any]) -> List[Dict[str, str]]:
    return [{"rule_id": r["id"], "target_queue": r["target_queue"]} for r in RULES if evaluate_condition(r["condition"], event_detail)]

def send_to_queue(queue_url: str, message_body: str, span: trace.Span) -> bool:
    for attempt in range(5):
        try:
            sqs_client.send_message(QueueUrl=queue_url, MessageBody=message_body, MessageGroupId="interaction-events", MessageDeduplicationId=str(os.urandom(8).hex()))
            span.set_attribute("sqs.send.success", True)
            return True
        except ClientError as e:
            if e.response["Error"]["Code"] not in ("ThrottlingException", "ServiceUnavailable"):
                return False
            if attempt < 4:
                time.sleep(0.5 * (2 ** attempt))
        except Exception:
            return False
    return False

def route_to_dlq(message: Dict[str, Any], error_reason: str) -> None:
    try:
        sqs_client.send_message(QueueUrl=DLQ_URL, MessageBody=json.dumps({"original_message": message.get("Body"), "error": error_reason}))
    except Exception as e:
        logger.error("DLQ send failed: %s", e)

def process_event(message: Dict[str, Any]) -> None:
    with tracer.start_as_current_span("event.process") as span:
        span.set_attribute("message.id", message.get("MessageId"))
        try:
            body = json.loads(message.get("Body", "{}"))
            detail = body.get("detail", {})
            if not detail:
                raise ValueError("Missing detail field")
            span.set_attribute("interaction.id", detail.get("id"))
            matched = match_rules(detail)
            span.set_attribute("rules.matched", len(matched))
            if not matched:
                sqs_client.delete_message(QueueUrl=SOURCE_QUEUE_URL, ReceiptHandle=message["ReceiptHandle"])
                return
            for target in matched:
                with tracer.start_as_current_span("sqs.fanout", attributes={"rule.id": target["rule_id"]}) as fspan:
                    if not send_to_queue(target["target_queue"], message["Body"], fspan):
                        fspan.set_status(trace.Status(trace.StatusCode.ERROR, "Fan-out failed"))
                        raise RuntimeError("Fan-out failed")
            sqs_client.delete_message(QueueUrl=SOURCE_QUEUE_URL, ReceiptHandle=message["ReceiptHandle"])
            span.set_attribute("processing.status", "success")
        except Exception as e:
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            span.record_exception(e)
            route_to_dlq(message, str(e))

def main() -> None:
    logger.info("Starting Genesys Cloud EventBridge consumer")
    while True:
        with tracer.start_as_current_span("sqs.receive_batch") as span:
            try:
                response = sqs_client.receive_message(QueueUrl=SOURCE_QUEUE_URL, MaxNumberOfMessages=10, WaitTimeSeconds=5)
                messages = response.get("Messages", [])
                span.set_attribute("messages.received", len(messages))
                for msg in messages:
                    process_event(msg)
            except ClientError as e:
                span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
                logger.error("SQS receive error: %s", e)
                time.sleep(2)
            except Exception as e:
                span.record_exception(e)
                logger.error("Unexpected error: %s", e)
                time.sleep(5)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden on SQS Operations

• Cause: The IAM role attached to the execution environment lacks sqs:ReceiveMessage, sqs:DeleteMessage, or sqs:SendMessage permissions for the specified queue ARNs.
• Fix: Update the IAM policy to include explicit permissions for the source queue, target queues, and DLQ. Verify the trust policy allows the service principal to assume the role.
• Code verification: Check AWS CloudTrail logs for UnauthorizedOperation and confirm the QueueUrl matches the ARN in the IAM policy.

Error: 429 ThrottlingException on Fan-Out

• Cause: Exceeding the per-second message rate limit on a target SQS queue (default is 3000 messages per second for standard queues, or 3000 with batching).
• Fix: The retry logic implements exponential backoff. Increase base_delay in send_to_queue or enable SQS long polling on the consumer side to reduce API call frequency. Consider batching messages if routing to the same queue.
• Code verification: Monitor sqs.send.exhausted_retries attributes in OpenTelemetry spans. Adjust max_retries if transient throttling persists during peak interaction volumes.

Error: 5xx ServiceUnavailable or InternalError

• Cause: Temporary AWS infrastructure degradation or EventBridge rule evaluation failures.
• Fix: The botocore retry configuration with mode: "adaptive" handles transient failures automatically. If errors persist, verify the EventBridge rule filter matches the Genesys Cloud payload structure exactly.
• Code verification: Check CloudWatch metrics for NumberOfMessagesSent and ApproximateAgeOfOldestMessage. A growing backlog indicates processing bottlenecks rather than API failures.

Error: Rule Evaluation Fails Silently

• Cause: JSONPath expressions reference nested fields that do not exist in the Genesys Cloud detail object, or the condition string contains syntax errors.
• Fix: Validate the Genesys Cloud interaction completion event schema. Use jsonpath-ng test cases before deployment. The evaluate_condition function catches JsonPathParserError and returns False, preventing crashes.
• Code verification: Enable debug logging to see Rule evaluation failed warnings. Compare the actual event payload against the detail.tags, detail.duration_ms, and detail.channel paths.

Official References

• Genesys Cloud EventBridge Integration API
• AWS SQS API Reference
• OpenTelemetry Python SDK Documentation
• Boto3 SQS Client Documentation
• CloudEvents Specification