Event Streaming (SSE) - ChatATP Studio Documentation

ChatATP uses Server-Sent Events (SSE) to stream real-time updates as agents process messages, execute tools, and generate responses.

SDK Developer Scenarios: Event Streams

The ChatATP SDK handles parsing the underlying HTTP stream. Here are the core scenarios you should handle when reading events from the chat_stream method.

Scenario 1: Basic Text Response

When a user asks a simple question and the agent responds without needing tools.

import sys

async for event in await client.chat_stream(agent_id=7, ...):
    if event.type == "conversation.message.created":
        pass # Initial acknowledgment
    elif event.type == "message.created":
        pass # User message saved
    elif event.type == "agent.response.delta":
        # Incremental text chunks from the LLM
        sys.stdout.write(event.data.get("delta", ""))
        sys.stdout.flush()
    elif event.type == "agent.response.completed":
        # Finalization of the agent's full response
        print(f"\nFinished text: {event.data.get('content')}")

Scenario 2: Autonomous Tool Execution

When the agent decides to invoke an internal tool during its response cycle.

async for event in await client.chat_stream(agent_id=7, ...):
    if event.type == "tool.execution.started":
        # 1. The agent decides to call a tool
        print(f"Running tool: {event.data.get('name')}", event.data.get("arguments"))
    elif event.type == "tool.execution.completed":
        # 2. The backend executes the tool and streams back the result
        print(f"Tool {event.data.get('name')} completed. OK: {event.data.get('ok')}")
        print(f"Result: {event.data.get('result')}")
    elif event.type == "agent.response.delta":
        # 3. The agent reasons about the tool response and streams text
        sys.stdout.write(event.data.get("delta", ""))
        sys.stdout.flush()
    elif event.type == "agent.response.completed":
        # 4. Final completion payload with full metadata
        print("\nResponse complete. Tools used:", event.data.get("tool_calls"))

Scenario 3: Error Encountered

When something goes wrong in the backend or tool execution fails.

async for event in await client.chat_stream(agent_id=7, ...):
    if event.type == "tool.execution.completed":
        if not event.data.get("ok"):
            print(f"Tool failed: {event.data.get('result')}")
    elif event.type == "error":
        # If the LLM integration completely drops or a critical pipeline error occurs
        print(f"Critical Error: {event.data.get('message')}")

SDK Integration Tips:

State Management: Keep a dictionary in memory keyed by tool_call_id. When tool.execution.started arrives, push the placeholder. When tool.execution.completed arrives, update the matching placeholder with the result string.
Text Accumulation: Simply concatenate the strings from agent.response.delta sequentially to render typewriter effects.