How to add Server-Sent Events (SSE) streaming to an A2A agent so consumers get intermediate progress updates instead of blocking for the full response.

Where streaming fits in the dispatch pipeline

All skill requests — whether triggered by a Discord DM, a Linear webhook, or an autonomous action — travel the same unified dispatch path:

inbound event
  → RouterPlugin
    → agent.skill.request (bus)
      → SkillDispatcherPlugin
        → ExecutorRegistry.resolve(skill, targets?)
          → A2AExecutor

Streaming is a transport detail at the A2AExecutor layer. If the agent's card declares streaming: true, the executor sends message/stream instead of message/send and reads the SSE response as it arrives. The caller — SkillDispatcherPlugin, the scheduled cron / ceremony loop, Ava — does not know or care whether the underlying call was streaming or blocking. The result arrives on replyTopic either way.

Overview

By default, A2A agents handle message/send — the executor blocks until the agent finishes. For long-running skills (deep research, multi-step triage), this creates dead air. SSE streaming solves this:

1. A2AExecutor sends message/stream instead of message/send
2. Agent returns text/event-stream with intermediate updates
3. Executor fires onStreamUpdate callbacks as events arrive (e.g. Discord o11y)
4. Agent sends final artifact + [DONE] to close the stream; executor publishes to replyTopic

Agent-side setup

1. Declare streaming in your agent card

AGENT_CARD = {
    "name": "my-agent",
    "capabilities": {"streaming": True},  # ← enables SSE
    # ...
}

2. Handle message/stream in your /a2a endpoint

Accept both message/send (blocking) and message/stream (SSE):

from fastapi.responses import StreamingResponse
import json, asyncio, queue

@app.post("/a2a")
async def a2a_handler(req: dict):
    method = req.get("method", "")

    if method == "message/stream":
        return StreamingResponse(
            sse_generator(req),
            media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
        )
    elif method == "message/send":
        return blocking_handler(req)
    else:
        return {"jsonrpc": "2.0", "error": {"code": -32601, "message": "Unknown method"}}

3. Emit SSE events

Each SSE line must be data: <json>\n\n. Three event types:

Critical — every event needs a kind discriminator. @a2a-js/sdk's for await loop routes by kind. Events without it are silently skipped and A2AExecutor sees the stream as empty. This is what Quinn #40 fixed. Wire fields are camelCase (taskId, contextId, lastChunk) — not snake_case.

Initial snapshot — the first frame is a full Task object with kind: "task":

data: {"kind": "task", "id": "task-uuid", "contextId": "conv-uuid", "status": {"state": "submitted", "timestamp": "..."}}

TaskStatusUpdateEvent — progress indicator:

data: {"kind": "status-update", "taskId": "task-uuid", "contextId": "conv-uuid", "status": {"state": "working", "message": {"role": "agent", "parts": [{"kind": "text", "text": "Scanning HuggingFace..."}]}}, "final": false}

TaskArtifactUpdateEvent — partial result:

data: {"kind": "artifact-update", "taskId": "task-uuid", "contextId": "conv-uuid", "artifact": {"artifactId": "task-uuid", "parts": [{"kind": "text", "text": "Found 3 relevant papers..."}]}, "append": true, "lastChunk": false}

Completion — emit TWO events: the final artifact-update, then a final status-update. [DONE] closes the stream:

data: {"kind": "artifact-update", "taskId": "task-uuid", "contextId": "conv-uuid", "artifact": {"artifactId": "task-uuid", "parts": [{"kind": "text", "text": "Full research report..."}]}, "append": false, "lastChunk": true}

data: {"kind": "status-update", "taskId": "task-uuid", "contextId": "conv-uuid", "status": {"state": "completed", "timestamp": "..."}, "final": true}

data: [DONE]

4. Example SSE generator

async def sse_generator(req):
    task_id = str(uuid.uuid4())
    context_id = req["params"].get("contextId", task_id)
    text = extract_text(req)

    # Frame 0: initial Task snapshot
    yield f"data: {json.dumps({'kind': 'task', 'id': task_id, 'contextId': context_id, 'status': {'state': 'submitted', 'timestamp': iso_now()}})}\n\n"

    # Run work with progress callback
    progress_q = queue.Queue()

    async def on_progress(msg):
        progress_q.put(msg)

    task = asyncio.create_task(do_work(text, on_progress))

    while not task.done():
        await asyncio.sleep(0.5)
        while not progress_q.empty():
            msg = progress_q.get_nowait()
            yield f"data: {json.dumps({'kind': 'status-update', 'taskId': task_id, 'contextId': context_id, 'status': {'state': 'working', 'message': {'role': 'agent', 'parts': [{'kind': 'text', 'text': msg}]}}, 'final': False})}\n\n"

    result = task.result()
    # Terminal: artifact-update FIRST (with the authoritative artifact),
    # then status-update with final=True.
    yield f"data: {json.dumps({'kind': 'artifact-update', 'taskId': task_id, 'contextId': context_id, 'artifact': {'artifactId': task_id, 'parts': [{'kind': 'text', 'text': result}]}, 'append': False, 'lastChunk': True})}\n\n"
    yield f"data: {json.dumps({'kind': 'status-update', 'taskId': task_id, 'contextId': context_id, 'status': {'state': 'completed', 'timestamp': iso_now()}, 'final': True})}\n\n"
    yield "data: [DONE]\n\n"

Workstacean-side setup

1. Declare streaming in agents.yaml

agents:
  - name: protopen
    url: "${PROTOPEN_BASE_URL}/a2a"
    streaming: true    # ← enables SSE client
    skills:
      - name: passive_recon

2. Discord o11y (optional)

Set DISCORD_AGENT_OPS_CHANNEL to a Discord channel ID. When streaming is active, intermediate SSE events are posted to this channel so you can watch agents think in real time.

The SkillBrokerPlugin wires the onStreamUpdate callback automatically when the channel is set.

How it works end-to-end

Inbound message: "run passive recon on example.com"
  → RouterPlugin → agent.skill.request (bus)
    → SkillDispatcherPlugin
      → ExecutorRegistry → A2AExecutor (streaming: true)
        → POST /a2a  method: message/stream
          ← text/event-stream:
              data: {"kind":"task", "id":"...", "contextId":"...", "status":{"state":"submitted"}}
              data: {"kind":"status-update", "taskId":"...", "status":{"state":"working","message":{...}}, "final":false}
              data: {"kind":"status-update", "taskId":"...", "status":{"state":"working","message":{...}}, "final":false}
              data: {"kind":"artifact-update", "taskId":"...", "artifact":{"artifactId":"...","parts":[...]}, "append":false, "lastChunk":true}
              data: {"kind":"status-update", "taskId":"...", "status":{"state":"completed"}, "final":true}
              data: [DONE]
        Each "working" event → onStreamUpdate → Discord agent-ops channel
        Final artifact → published to replyTopic

The caller receives the same SkillResult it would have received from a blocking call. Streaming is invisible above the executor layer.

Task lifecycle states

State	Meaning	Stream behavior
working	Agent processing	Emitted as progress updates
input-required	Agent needs human input	Stream pauses; HITL gate raised
completed	Done	Final artifact emitted, stream closes
failed	Error	Error message emitted, stream closes

input-required and HITL

If a streaming agent returns input-required, the task does not complete. TaskTracker detects the state and terminates the task as failed with a "no approval gate is wired" message — this build doesn't ship an in-process HITL plugin. If your deployment needs human approval mid-flow, register a bus subscriber for the relevant prompt topic and publish the response back yourself.

Fallback behavior

If the agent doesn't support streaming (card says streaming: false or omits it), A2AExecutor falls back to blocking message/send. No code changes needed on the caller side.

If the agent supports streaming but the SSE connection fails, the executor catches the error and falls back to message/send automatically.

Artifact chunking

Agents can progressively stream a long artifact in multiple frames using append + lastChunk on TaskArtifactUpdateEvent:

• append: false (or omitted) → the incoming parts replace the buffer for that artifactId
• append: true → the incoming parts are concatenated to the existing buffer
• lastChunk: true → finalize the artifact; A2AExecutor fires onStreamUpdate({ type: "artifact_complete", ... })

This lets a researcher stream a 2000-word report as 20 × 100-word chunks. Each chunk surfaces live via onStreamUpdate({ type: "artifact_chunk", ... }) to Discord, the dashboard, etc. The final concatenated text is what lands in the SkillResult.text.

Long-running tasks

If your agent returns a non-terminal task (working, submitted, input-required) rather than a final result, TaskTracker polls tasks/get (or uses tasks/resubscribe for streaming agents) and publishes the response on the original replyTopic once terminal. No caller-side timeout tuning needed.

Streaming progress out of workstacean (inbound A2A clients)

Everything above is about how A2AExecutor consumes a remote agent's stream. The reverse direction — when an external A2A client calls workstacean's /a2a endpoint and we want to surface intermediate progress to that caller — is wired through the bus with a single topic family.

Topic contract

agent.skill.progress.{correlationId}

BusAgentExecutor (the A2A server adapter) subscribes to this topic on every inbound message/send or message/stream call alongside the reply topic. Each event published there is translated into an A2A status-update with state: "working" and final: false, sent over the SSE channel.

Payload shape:

interface AgentSkillProgressPayload {
  text?: string;                  // visible message body in the streamed update
  percent?: number;               // 0-100 progress
  step?: string;                  // named phase the executor is in
  meta?: Record<string, unknown>; // arbitrary structured progress
}

All fields are optional. The bus consumer only attaches a message body to the streamed status-update when text is set; percent / step / meta go into status.metadata for clients that render progress affordances rather than text.

Opt-in publishing from a skill executor

A skill executor that wants to stream progress publishes during work:

bus.publish(`agent.skill.progress.${req.correlationId}`, {
  id: crypto.randomUUID(),
  correlationId: req.correlationId,
  topic: `agent.skill.progress.${req.correlationId}`,
  timestamp: Date.now(),
  payload: { text: "Searching project history…", percent: 25, step: "search" },
});

Executors that don't publish to this topic lose nothing — the existing submitted → working → completed lifecycle still works as before.

Late-event safety

Progress events that arrive after the terminal reply (race with an unusually fast executor) are silently dropped — BusAgentExecutor tracks the settled state and unsubscribes both agent.skill.progress.* and agent.skill.response.* together on terminal / cancel. No leaked subscriptions across overlapping task ids.

Consumer expectations

A streaming A2A client subscribed to message/stream previously saw:

submitted → working → ··· silence ··· → completed (full text)

Now sees (when the executor emits progress):

submitted → working → working (text="...") → working (percent=40, step="...") → ··· → completed

Same task lifecycle, real-time visibility into what the agent is doing.

Structured tool-call frames (tool-call-v1)

The agent.skill.progress.{correlationId} channel above carries humanized text — a one-liner like routing to quinn or searching the web on status.message, which simple clients render as-is. Alongside it, the in-process runtime emits a structured tool timeline on a parallel channel for rich clients that want to draw a per-tool spinner with started → completed/failed state.

Topic contract

agent.skill.toolframe.{correlationId}

BusAgentExecutor subscribes to this topic on every inbound message/send / message/stream call, alongside the reply and progress topics. Each event payload is { frame: ToolCallArtifactData }, and each frame becomes its own A2A artifact-update event whose artifact (named tool-call) carries a tool-call-v1 DataPart:

eventBus.publish(AgentEvent.artifactUpdate({
  taskId, contextId,
  artifact: dataArtifact([emitToolCall(frame)], { name: "tool-call" }),
  append: false,
  lastChunk: false,
}));

The frame is a discriminated union on phase — { toolCallId, name, phase: "started", args? }, { ..., phase: "completed", result? }, or { ..., phase: "failed", error? }. A client keys on toolCallId to stitch each started to its later completed / failed. See the tool-call-v1 extension reference for the full payload shape and emit/parse helpers.

Where the frames come from

DeepAgentExecutor streams its LangGraph agent (streamMode: "values") and, as each tool turn streams in, fires an onToolFrame callback (a started frame per AI-message tool call keyed by the call id; a completed / failed frame per tool-result message). AgentRuntimePlugin bridges that callback to agent.skill.toolframe.{correlationId}, and BusAgentExecutor emits each as the artifact-update above.

The initial "thinking" frame

The first LLM turn can take 15–20s before it produces any tool call, so the runtime also emits an immediate thinking progress frame the instant the graph starts — published on the text channel (agent.skill.progress.{cid}, step: "thinking") so even a simple client sees motion before any tool runs, instead of a byte-silent stream broken only by keepalive heartbeats.

Two channels, pick by client

Channel	Topic	A2A surface	Reader
Text narration	agent.skill.progress.{cid}	status-update (status.message)	Simple clients
Structured frames	agent.skill.toolframe.{cid}	artifact-update DataPart	Rich tool-timeline clients

Both ride the same SSE stream and both are torn down together with the reply subscription on terminal / cancel; late frames after settle are dropped. A simple client reads status.message and ignores the artifact frames; a rich client reads the frames to render a live tool timeline.

Push-notification configs survive restart

Inbound A2A clients can register a webhook callback via tasks/pushNotificationConfig/set so workstacean POSTs them when long-running work terminates. Those configs are persisted to ${DATA_DIR}/push-notifications.db (SQLite, WAL journal) so they survive container restarts — important on :main where watchtower auto-pulls multiple times per day. Each config carries a 24-hour TTL by default; expired rows are filtered from load() and opportunistically purged on subsequent save() calls. Falls back to the SDK's in-memory store when DATA_DIR is unset (tests + ephemeral setups).