amazon-bedrock-agentcore-samples/02-use-cases/SRE-agent/gateway/observability.py

import boto3


def enable_observability_for_resource(
    resource_arn: str,
    resource_id: str,
    account_id: str,
    region: str = 'us-east-1'
):
    """
    Enable observability for a Bedrock AgentCore resource (e.g., Memory Store)
    """
    logs_client = boto3.client('logs', region_name=region)

    # Step 0: Create new log group for vended log delivery
    log_group_name = f'/aws/vendedlogs/bedrock-agentcore/{resource_id}'
    logs_client.create_log_group(logGroupName=log_group_name)
    log_group_arn = f'arn:aws:logs:{region}:{account_id}:log-group:{log_group_name}'
    
    # Step 1: Create delivery source for logs
    logs_source_response = logs_client.put_delivery_source(
        name=f"{resource_id}-logs-source",
        logType="APPLICATION_LOGS",
        resourceArn=resource_arn
    )
    
    # Step 2: Create delivery source for traces  
    traces_source_response = logs_client.put_delivery_source(
        name=f"{resource_id}-traces-source", 
        logType="TRACES",
        resourceArn=resource_arn
    )
    
    # Step 3: Create delivery destinations
    logs_destination_response = logs_client.put_delivery_destination(
        name=f"{resource_id}-logs-destination",
        deliveryDestinationType='CWL',
        deliveryDestinationConfiguration={
            'destinationResourceArn': log_group_arn,
        }
    )
    
    # Traces required for memory only
    traces_destination_response = logs_client.put_delivery_destination(
        name=f"{resource_id}-traces-destination",
        deliveryDestinationType='XRAY'
    )
    
    # Step 4: Create deliveries (connect sources to destinations)
    logs_delivery = logs_client.create_delivery(
        deliverySourceName=logs_source_response['deliverySource']['name'],
        deliveryDestinationArn=logs_destination_response['deliveryDestination']['arn']
    )
    
    # Traces required for memory only
    traces_delivery = logs_client.create_delivery(
        deliverySourceName=traces_source_response['deliverySource']['name'], 
        deliveryDestinationArn=traces_destination_response['deliveryDestination']['arn']
    )
    
    print(f"Observability enabled for {resource_id}")
    return {
        'logs_delivery_id': logs_delivery['id'],
        'traces_delivery_id': traces_delivery['id']
    }