Events / Streaming Guide

This guide explains how to use the SDK's Events module to deliver agent execution state to the frontend in real time.

Overview

The Events module is an asynchronous event delivery system centered around EventEmitter. Events emitted by the agent can be simultaneously delivered to three destinations: SSE (Server-Sent Events), Webhook, and DB.

Agent → EventEmitter → ┬→ SSE Handler     → Frontend
                        ├→ DB Handler      → PostgreSQL
                        └→ Webhook Handler → External System

Event Types

A list of event types that agents can emit.

About the "UI Display" column

Rendering follows the platform ChatUI. A dedicated UI requires the right metadata — see the implementation guide by display method for what each one needs.

EventType	Purpose	UI Display
`PHASE_START`	Phase start	Phase name
`PROGRESS_UPDATE`	Progress update	Progress message
`PROGRESS_MESSAGE`	Processing notice	Loading indicator
`THOUGHT_MESSAGE`	Agent thought process	Thought bubble
`TOOL_START`	Tool execution start	Task card
`TOOL_RESULT`	Tool execution complete	Task card (updated)
`COMPLETION_SUCCESS`	Processing complete (success)	Completion notification
`COMPLETION_FAILURE`	Processing complete (failure)	Error message
`FILE_CREATED`	File generation complete	Download link
`HITL_REQUIRED_BROWSER_VNC`	Human intervention request for browser operation	VNC connection UI
`HITL_REQUIRED_BROWSER_CLI`	Human intervention request for CLI operation	Message
`HITL_COMPLETED`	Human intervention complete	Completion notification
`PERMISSION_REQUEST`	Execution permission request	Approval dialog
`PERMISSION_RESPONSE`	Execution permission response	Approval result
`USER_INTERACTION_REQUIRED`	User input request	Choice buttons
`UNEXPECTED_ERROR`	Unexpected error	Error notification

Implementation Guide by Display Method

This section groups the implementation needed to actually get an event on screen, by how it displays.

Displayed with `message` alone (no `metadata`)

Just emit with message and it displays. Applies to: PHASE_START / PROGRESS_UPDATE / COMPLETION_SUCCESS / UNEXPECTED_ERROR / PERMISSION_RESPONSE / HITL_REQUIRED_BROWSER_CLI

How each one renders differs per event — see the "UI Display" column above. To show that work is in progress, use PROGRESS_MESSAGE (loading indicator).

Rendered as a thought bubble

This one needs no metadata either — just emit with message — but it renders in the reasoning display area instead. Applies to: THOUGHT_MESSAGE

Rendered as a task card (tool execution)

TOOL_START / TOOL_RESULT render as task cards, drawn from the flattened metadata. title is required — it becomes the card's heading.

metadata key	Purpose
`title`	Required. The task card heading
`description`	Supplementary text on the card
`status`	`in_progress` (clock icon) / `completed` (check) / `error` (cross)
`actionType`	Selects the dedicated renderer (below). If omitted you still get the `title` / `description` fallback
`files[]`	Attached files, shaped `{id, filename, fileType, size, url}`
`metadata.call_id`	Update key for one task. Pass the same value on `TOOL_START` and `TOOL_RESULT` to update a single card

Accepted actionType values: command_execution / code_execution / file_operation / browser_action / search_result / image_search / markdown_display

Python
# Tool start — emit one card with the clock icon
await emitter.emit_event(
    event_type=EventType.TOOL_START,
    message="Running a command",
    metadata={
        "title": "Command execution",
        "status": "in_progress",
        "actionType": "command_execution",
        "metadata": {"call_id": "call-1", "command": "ls -la"},
    },
)

# Tool result — update the card with the same call_id to completed
await emitter.emit_event(
    event_type=EventType.TOOL_RESULT,
    message="Command finished",
    metadata={
        "title": "Command execution",
        "status": "completed",
        "actionType": "command_execution",
        "metadata": {"call_id": "call-1", "command": "ls -la", "output": "..."},
    },
)

Rendered as files

Use metadata.files[] for in-progress files and metadata.deliverable_files[] for deliverables in the completion body (file_name / download_url are not used). Each entry in files[] is shaped {id, filename, fileType, size, url}. See the storage guide for the exact shape. Applies to: FILE_CREATED / COMPLETION_SUCCESS

Rendered as choice / confirmation buttons

Add metadata.interaction_type and a button UI appears under the message.

"choice" + options[] → choice buttons; pressing one submits that value as-is. options accepts at most 10 entries, strings only, each at most 200 characters. Applies to: USER_INTERACTION_REQUIRED
"confirmation" (no options) → yes / no buttons. Applies to: PERMISSION_REQUEST

Python
# Offer choices and let the user pick
await emitter.emit_event(
    event_type=EventType.USER_INTERACTION_REQUIRED,
    message="Which output format do you want?",
    metadata={"interaction_type": "choice", "options": ["PDF", "Excel", "Markdown"]},
)

# Ask for approval with yes / no
await emitter.emit_event(
    event_type=EventType.PERMISSION_REQUEST,
    message="May I delete this file?",
    metadata={"interaction_type": "confirmation"},
)

Rendered as a loading indicator

Emit PROGRESS_MESSAGE to show the animated loading indicator.

Python
await emitter.emit_event(
    event_type=EventType.PROGRESS_MESSAGE,
    message="Searching internal documents...",
)

Rendered as the VNC connection UI

For HITL_REQUIRED_BROWSER_VNC / HITL_COMPLETED, put the following keys into metadata as-is.

metadata key	Purpose
`type`	`"vnc_popup"` (intervention required) / `"vnc_completed"` (done)
`execution_id`	The execution this refers to
`vnc_url`	VNC URL to connect to
`cancel_url`	URL that cancels the intervention
`vnc_port`	VNC port
`intervention_type`	Kind of intervention
`instructions`	What the user should do
`reason`	Why the intervention became necessary

Python
await emitter.emit_event(
    event_type=EventType.HITL_REQUIRED_BROWSER_VNC,
    message="Please complete the login",
    metadata={
        "type": "vnc_popup",
        "execution_id": os.environ["EXECUTION_ID"],
        "vnc_url": vnc_url,
        "cancel_url": cancel_url,
        "vnc_port": 5901,
        "intervention_type": "browser_login",
        "instructions": "Log in on the screen that opens, then close it.",
        "reason": "Reached a page that requires authentication",
    },
)

Closing the turn

Always put the text you want displayed into the completion event's message. Close the turn — on success and on failure alike — by emitting COMPLETION_SUCCESS (finish_reason=stop) as the terminal event. When reporting a failure, include the failure details in message / metadata (the execution's own outcome is recorded separately via PodRuntime.final(status=...)).

COMPLETION_FAILURE does not close the turn. Put the final response in the COMPLETION_SUCCESS body.

Python
try:
    await run_agent_logic()
except Exception as e:
    # Close the turn with COMPLETION_SUCCESS (finish_reason=stop) even on failure.
    await emitter.emit_event(
        event_type=EventType.COMPLETION_SUCCESS,
        message=f"An error occurred: {e}",
        metadata={"error_type": type(e).__name__, "status": "failed"},
    )
finally:
    # Always execute cleanup
    await emitter.cleanup()

Basic Usage

Initializing and Using EventEmitter

emit_event() accepts message (string) and metadata (dictionary, optional).

Python
from agenticstar_platform.events import EventEmitter, EventType, SubEventType

# Initialize EventEmitter (execution_id is required)
emitter = EventEmitter(execution_id="exec-abc-123")

# Phase start event
await emitter.emit_event(
    event_type=EventType.PHASE_START,
    message="Analyzing intent...",
    metadata={"phase": "intent"},
)

# Streaming thought process
await emitter.emit_event(
    event_type=EventType.THOUGHT_MESSAGE,
    message="The user is requesting a RAG search",
    sub_event_type=SubEventType.SEARCH_WEB,
)

# Tool execution
await emitter.emit_event(
    event_type=EventType.TOOL_START,
    message="Executing search_knowledge",
    metadata={"tool_name": "search_knowledge", "input": {"query": "AI agent"}},
)

# Tool result
await emitter.emit_event(
    event_type=EventType.TOOL_RESULT,
    message="Search complete: 5 results",
    metadata={"tool_name": "search_knowledge", "result_count": 5},
)

# Completion
await emitter.emit_event(
    event_type=EventType.COMPLETION_SUCCESS,
    message="Response complete",
)

# Cleanup (required)
await emitter.cleanup()

For the terminal-event contract, see closing the turn.

Automatic Sequence Number Management

EventEmitter automatically assigns sequence numbers to emitted events. The frontend can use these numbers to guarantee event ordering.

Python
# Sequence number is available as a property
print(emitter.sequence_number)  # 0, 1, 2, ... auto-incremented

Event Handlers

EventEmitter accepts a single handler in its constructor.

SSE Handler (Frontend Delivery)

Python
from agenticstar_platform.events import EventEmitter, create_sse_handler

# Combine with FastAPI's StreamingResponse
async def chat_stream(request: ChatRequest):
    sse_handler = create_sse_handler()
    emitter = EventEmitter(execution_id="exec-123", handler=sse_handler)

    # Start agent processing asynchronously
    asyncio.create_task(run_agent(emitter, request))

    # Return SSE stream
    return StreamingResponse(
        emitter.consume_events(),
        media_type="text/event-stream",
    )

DB Handler (Persistence)

Python
from agenticstar_platform.events.handlers import DatabaseEventHandler

# Persist events to PostgreSQL with DB handler
db_handler = DatabaseEventHandler(
    data_access=data_access,
    user_id="user-001",
    conversation_id="conv-001",
    message_id="msg-001",
)
emitter = EventEmitter(execution_id="exec-123", handler=db_handler)

Webhook Handler (External Notification)

Python
from agenticstar_platform.events.handlers import WebhookEventHandler

# Send webhook notifications to external systems
webhook_handler = WebhookEventHandler(
    webhook_url="https://your-system.example.com/webhook",
    conversation_id="conv-001",
    message_id="msg-001",
)
emitter = EventEmitter(execution_id="exec-123", handler=webhook_handler)

Composite Handler

Multiple handlers can be combined and managed as one.

Python
from agenticstar_platform.events.handlers import CompositeEventHandler

composite = CompositeEventHandler([
    create_sse_handler(),
    DatabaseEventHandler(data_access=da, user_id="u", conversation_id="c", message_id="m"),
    WebhookEventHandler(webhook_url="https://...", conversation_id="c", message_id="m"),
])
emitter = EventEmitter(execution_id="exec-123", handler=composite)

SubEventType and `actionType`

sub_event_type alone does not change the rendering. To vary how a tool execution renders, set metadata.actionType using the table below. sub_event_type remains useful as a classification in the DB and in logs.

SubEventType → actionType mapping

SubEventType	metadata.actionType
SEARCH_WEB	`search_result`
COMMAND_EXECUTION / BASH_EXECUTED	`command_execution`
FILE_OPERATION / FILE_EDITED / FILE_READ	`file_operation`
IMAGE_GENERATED	`image_search` (image preview family)
Anything else	Leave it unset (`title` / `description` fallback) or pick the closest value

Python
# metadata.actionType is what varies the rendering; sub_event_type is for classification
await emitter.emit_event(
    event_type=EventType.TOOL_START,
    message="Executing web search...",
    sub_event_type=SubEventType.SEARCH_WEB,
    metadata={
        "title": "Web search",
        "status": "in_progress",
        "actionType": "search_result",
        "metadata": {"call_id": "call-2", "query": "AI agent"},
    },
)

SubEventType reference

SubEventType	Purpose
`SEARCH_WEB`	Web search in progress
`COMMAND_EXECUTION`	Command execution in progress
`FILE_OPERATION`	File operation in progress
`FILE_EDITED`	File edit complete
`FILE_READ`	File read complete
`FILE_SEARCHED`	File search complete
`BASH_EXECUTED`	Bash command execution complete
`WEB_FETCHED`	Web page fetch complete
`MCP_TOOL`	MCP tool execution in progress
`LOCAL_ASSISTANT`	Local assistant processing
`TASK_LAUNCHED`	Task launched
`TODO_UPDATED`	TODO updated
`VIDEO_GENERATED`	Video generation complete
`IMAGE_GENERATED`	Image generation complete
`SLIDE_CREATED`	Slide creation complete
`MACOS_AUTOMATION`	macOS automation in progress

Marketplace Handler

For agents provided through the marketplace, use the dedicated handler factory. It generates a composite handler for DB + Webhook in one step.

Python
import os

from agenticstar_platform.events import EventEmitter
from agenticstar_platform.events.handlers import create_marketplace_handler

# Read runtime identifiers from the environment variables injected by the platform
handler = create_marketplace_handler(
    data_access=data_access,
    webhook_url="https://tenant.example.com/webhook",
    user_id=os.environ["USER_ID"],
    conversation_id=os.environ["CONVERSATION_ID"],
    message_id=os.environ["MESSAGE_ID"],
    request_source="external",  # Set when delivering responses via the external API
)
emitter = EventEmitter(execution_id=os.environ["EXECUTION_ID"], handler=handler)

SDK 0.5.29+

The whole execution lifecycle — this handler wiring, identity handoff, the terminal event (exactly once), and cleanup — is handled in one call by run_marketplace_agent(my_agent) (see the runner section in the quickstart). Using create_marketplace_handler directly is the low-level API for when you want to compose the handler yourself.

Error Handling

Python
try:
    await run_agent_logic()
except Exception as e:
    # Emit error event to notify the frontend
    await emitter.emit_event(
        event_type=EventType.COMPLETION_FAILURE,
        message=str(e),
        metadata={"error_type": type(e).__name__},
    )
finally:
    # Always execute cleanup
    await emitter.cleanup()

Next Steps

SDK API Reference — Events Module

Complete specifications for EventEmitter / StreamingEvent / EventType

View guide

Architecture Guide

SDK module structure and design philosophy

View guide

Overview​

Event Types​

Implementation Guide by Display Method​

Displayed with message alone (no metadata)​

Rendered as a thought bubble​

Rendered as a task card (tool execution)​

Rendered as files​

Rendered as choice / confirmation buttons​

Rendered as a loading indicator​

Rendered as the VNC connection UI​

Closing the turn​

Basic Usage​

Initializing and Using EventEmitter​

Automatic Sequence Number Management​

Event Handlers​

SSE Handler (Frontend Delivery)​

DB Handler (Persistence)​

Webhook Handler (External Notification)​

Composite Handler​

SubEventType and actionType​

SubEventType → actionType mapping​

SubEventType reference​

Marketplace Handler​

Error Handling​

Next Steps​