Skip to main content
Sometimes you might want to extract insights from the full conversation transcript after it has completed to analyze patterns, to extract key information, or to generate summaries for further processing.

​
Procedural agents for automated analysis

Operator has a built-in system for post-conversation analysis through procedural agents. These are specialized agents that can be configured to automatically run after a conversation ends, using an LLM to extract insights and generate outputs that can be fetched programmatically or delivered via a webhook. Procedural agents are triggered automatically when a conversation reaches completion. They:
  • Receive the full conversation context including all messages, tool calls, and events
  • Run a multi-step analysis prompts using an LLM, call tools connected to Operator
  • Generate structured outputs (summaries, extracted data, classifications, etc.)
  • Store results that can be retrieved via API or delivered via a webhook
See more at Procedural Agents.

​
Manual transcript assembly via API

For cases where you need custom processing or more control over the transcript assembly, you can manually fetch and process conversation data through the API. Note that this example focuses on assembling completed conversations for analysis. For rendering ongoing conversations with real-time message streaming, see our Real-time events example. When manually assembling conversation transcripts:
  • Focus on completed events: Use agent.message.completed and agent.tool_call.returned rather than delta/streaming events for clean final content
  • The data format is SSE, with JSON inside: Either use an SSE client and parse the JSON payloads or manually parse out individual blocks and the data: lines that compose the JSON payload
The following examples show how to assemble user messages, final agent messages, tool calls, events, and notices from a completed conversation.
  • python
  • node.js
# Before running, install the SSEClient package: pip install sseclient-py
# Or run the script with uv: uv run --with sseclient-py ./script.py
import requests
import json
import sseclient

def fetch_conversation_transcript(conversation_id, api_key):
    headers = {
        'Authorization': f'Bearer {api_key}',
        'Operator-Version': '2025-06-19',
        'Accept': 'text/event-stream'
    }

    # Fetch conversation events via SSE stream
    events_url = f'https://api.operator.xyz/conversations/{conversation_id}/events'
    response = requests.get(events_url, headers=headers, stream=True)
    response.raise_for_status()

    transcript = {
        'conversation_id': conversation_id,
        'messages': [],
        'tool_calls': [],
        'events': [],
        'notices': []
    }

    # Process SSE events to build transcript
    client = sseclient.SSEClient(response)
    for sse_event in client.events():
        if sse_event.data:
            event = json.loads(sse_event.data)
            event_type = event.get('payload', {}).get('type')

            # User messages
            if event_type == 'user.message.received':
                transcript['messages'].append({
                    'role': 'user',
                    'content': event['payload']['text'],
                    'timestamp': event['created_at']
                })

            # Final agent messages (completed only)
            elif event_type == 'agent.message.completed':
                transcript['messages'].append({
                    'role': 'agent',
                    'content': event['payload']['text'],
                    'generation_id': event['payload']['generation_id'],
                    'timestamp': event['created_at']
                })

            # Tool calls (completed only)
            elif event_type == 'agent.tool_call.returned':
                transcript['tool_calls'].append({
                    'call_id': event['payload']['call_id'],
                    'name': event['payload']['name'],
                    'arguments': event['payload']['arguments'],
                    'result': event['payload'].get('result'),
                    'error': event['payload'].get('error'),
                    'timestamp': event['created_at']
                })

            # Conversation events
            elif event_type in ['conversation.started', 'conversation.ended']:
                transcript['events'].append({
                    'type': event_type,
                    'timestamp': event['created_at']
                })

            # Server notices and errors
            elif event_type == 'server.error':
                transcript['notices'].append({
                    'type': 'error',
                    'message': event['payload']['error'],
                    'timestamp': event['created_at']
                })

    return transcript

conversation_id = 'your-conversation-id'
api_key = 'your-api-key'

transcript = fetch_conversation_transcript(conversation_id, api_key)

print("\nConversation Transcript:")

for message in transcript['messages']:
    if message['role'] == 'user':
        print(f"User: {message['content']}")
    elif message['role'] == 'agent':
        print(f"Agent: {message['content']}")

for tool_call in transcript['tool_calls']:
    print(f"Tool: {tool_call['name']} - {tool_call.get('result', tool_call.get('error', 'No result'))}")

for notice in transcript['notices']:
    if notice['type'] == 'error':
        print(f"Error: {notice['message']}")
⌘I