SSE API

This page documents MobKit v0.8.34 (mirrored from v0.8.34). MobKit uses Server-Sent Events (SSE) for several live HTTP streams:

agent-scoped streaming through GET /agents/{agent_id}/events
mob-merged streaming through GET /mob/events
structural mob events through GET /mobkit/mob_events/stream
lossy WorkGraph wake events through GET /mobkit/workgraph/facts/stream
console timeline replay and live continuation through GET /console/timeline/stream
identity-filtered console timeline streaming through GET /console/identity/{identity}/stream

The related mobkit/events/subscribe JSON-RPC method returns bounded replay snapshots in SSE frame format, but does not open a stream.

Authentication

Every SSE route honors RuntimeDecisionState.console.require_app_auth. When require_app_auth is on, the request must carry one of:

Authorization: Bearer <token> header, or
?auth_token=<token> query parameter

Missing or invalid credentials return HTTP 401. Tokens are percent-decoded; a token containing characters that need URL-encoding (=, &, +, %) is handled correctly.

Agent event stream

GET /agents/{agent_id}/events

Subscribe to raw Meerkat agent events for one agent (Content-Type: text/event-stream).

Frame format

id: worker-1:0
event: text_delta
data: {"type": "text_delta", "delta": "Analyzing the metrics now..."}

id: worker-1:1
event: tool_call_requested
data: {"type": "tool_call_requested", "id": "tool-1", "tool_call_id": "tool-1", "name": "query_metrics", "args": {"metric": "latency_p95"}}

id: worker-1:2
event: run_completed
data: {"type": "run_completed", "session_id": "session-1", "result": "Done."}

Console timeline stream

GET /console/timeline/stream

Streams a bounded replay snapshot followed by live console timeline frames. The route accepts the same identity, conversation_id, after, before, mode, and limit parameters as GET /console/timeline. If replay cannot satisfy the requested cursor, the server returns HTTP 409 with:

{
  "error": "replay_unavailable",
  "requested_cursor": "console:10",
  "latest_cursor": "console:240"
}

Clients should resume from latest_cursor rather than retrying the stale cursor.

Identity timeline stream

GET /console/identity/{identity}/stream

Convenience route for an identity-filtered console timeline stream. It sets the timeline query identity from the path and otherwise behaves like GET /console/timeline/stream.

Event subscription

Use the mobkit/events/subscribe JSON-RPC method to request a bounded point-in-time snapshot. It returns each selected event both as a typed envelope and as an SSE-ready frame. It does not open a live connection or return a stream URL; use the HTTP routes on this page for live observation. Example request:

{
  "jsonrpc": "2.0",
  "id": "sub-1",
  "method": "mobkit/events/subscribe",
  "params": {
    "scope": "agent",
    "agent_id": "worker-1",
    "last_event_id": "evt-42"
  }
}

The response includes:

Field	Description
`scope`	Normalized selected scope
`replay_from_event_id`	Requested checkpoint, or `null`
`keep_alive.interval_ms`	Server keep-alive cadence
`keep_alive.event`	Keep-alive event name
`keep_alive_comment`	SSE comment frame for keep-alive output
`event_frames`	SSE-ready frames for replay
`events`	Typed `EventEnvelope<UnifiedEvent>` values

Supported scopes:

Scope	Notes
`mob`	All merged module and agent events; this is the default
`agent`	Agent events only; requires a non-empty `agent_id`
`interaction`	Agent or module events whose `event_type` starts with `interaction`

For agent scope, agent_id may be the exact projected ID or the bare alias whose only difference is a trailing numeric generation suffix. The server applies the scope first and retains at most its three newest matching events. last_event_id is an event_id from an earlier mobkit/events/subscribe result, not an HTTP SSE frame ID. It must be present in that bounded, scope-filtered window. Replay is inclusive: the checkpoint event is returned again, followed by later retained events. An unknown, out-of-scope, or expired checkpoint returns JSON-RPC -32602, so reconnecting consumers should deduplicate the first event by event_id.

WorkGraph wake stream

GET /mobkit/workgraph/facts/stream

This route is present when the runtime has a WorkGraph service. It is a dedicated, lossy wake channel, not another view of the merged event log. Its data field is a bare UnifiedEvent::Module value with module mobkit.workgraph; the event is not wrapped in EventEnvelope and is not duplicated into GET /mob/events or mobkit/events/subscribe. Every connection starts with this mandatory resynchronization signal:

event: workgraph.resync_required
data: {"kind":"module","module":"mobkit.workgraph","event_type":"workgraph.resync_required","payload":{"reason":"initial_sync","skipped":null,"authority":"durable_workgraph_pull"}}

On receipt, read authoritative state through mobkit/workgraph/get, mobkit/workgraph/list, mobkit/workgraph/ready, or mobkit/workgraph/snapshot. Subsequent workgraph.fact events carry identifiers that can accelerate another read, but they are never state authority. The stream has no SSE id, Last-Event-ID, cursor query, catch-up, or replay contract. A slow consumer can miss events. When broadcast lag occurs, the server sends workgraph.resync_required with reason: "lagged" and the number of skipped wakes in skipped; read authoritative WorkGraph state again. When access control is enforced, the caller also needs the workgraph.view grant.

Structural mob events stream

GET /mobkit/mob_events/stream?after_seq=<cursor>&mob_id=<id>&run_id=<id>&event_types=<list>

Per-client SSE stream of structural MobEvents (typed envelopes preserving mob_id, run_id, step_id, agent_identity, and the full MobEventKind payload). Each connection opens its own MobEventsView::subscribe_after so catch-up and live tail share the same ordered stream — there is no race window between snapshot and live subscription.

Query parameters

Param	Type	Notes
`after_seq`	u64	Resume from this cursor (exclusive). When omitted, starts at the current `latest_cursor` (live tail only).
`mob_id`	string	Filter by mob id
`run_id`	string	Filter by flow run id
`step_id`	string	Filter by flow step id
`event_types`	comma-separated	e.g. `flow_started,step_completed`
`since_ms` / `until_ms`	u64	Inclusive / exclusive timestamp filters
`identity` / `member_id`	string	Filter by agent identity

Frame format

event: flow_started
id: mob-evt-1234
data: {"event_id":"mob-evt-1234","cursor":1234,"mob_id":"home","run_id":"...","kind":"flow_started",...}

Stale cursor

If after_seq > latest_cursor, the route returns HTTP 410 Gone with {"error":"event_query_stale","after_cursor":N,"latest_cursor":M}. SDK clients should rewind to latest_cursor and reconnect. The JSON-RPC handshake returns a snapshot frame plus a subscribe_url field that already encodes the resume cursor and filters:

{
  "stream": "mob_events",
  "events": [...],
  "next_after_seq": 1234,
  "subscribe_url": "/mobkit/mob_events/stream?after_seq=1234&mob_id=home",
  "keep_alive": {"interval_ms": 15000, "event": "keep_alive"}
}

after_seq falls back to latest_cursor captured at handshake when the snapshot is empty, so the SSE handler picks up gaplessly even when no events match yet.

Cursor durability

MobStructuralEvent.cursor is the upstream meerkat ledger cursor. Consumers can checkpoint it and reconnect with after_seq=<checkpointed cursor> only while the configured mob-event store retains that cursor and its later events. .persistent_state(path) alone does not make the mob-event ledger durable. It persists MobKit projection metadata, while the stock unified-runtime builder still uses in-memory Mob storage. Restart-safe replay requires a separately configured upstream Meerkat mob-event store that retains the ledger across the restart. If a checkpoint is ahead of the available ledger, the route returns HTTP 410 Gone as described above.

Keep-alive

The server sends keep-alive frames at a configurable interval (default: 15 seconds) to prevent connection timeouts:

: keep-alive

When the console timeline broadcast lags (slow consumer, channel saturated), the stream closes. Clients should refetch a recent timeline page and reconnect from the returned latest_cursor or a fresh Last-Event-ID.

Sending messages

To send a message to an agent, use the mobkit/send_message RPC method and capture the returned session_id for correlation:

{
  "accepted": true,
  "member_id": "lead-1",
  "session_id": "0195c3f2-8f40-7f95-a5a7-cc4f7d90d3d1"
}

Then observe live responses via GET /console/timeline/stream or GET /console/identity/{identity}/stream. Use mobkit/events/subscribe only when a bounded merged-event replay snapshot is sufficient.

Error handling

If the target agent/member does not exist or a subscription request is malformed, SSE routes return an HTTP error response before opening the stream:

Status	Cause
`400`	Empty or invalid member ID / malformed body
`404`	Member not found
`500`	Internal server error

Protocols

Authentication

Agent event stream

Frame format

Console timeline stream

Identity timeline stream

Event subscription

WorkGraph wake stream

Structural mob events stream

Query parameters

Frame format

Stale cursor

Cursor durability

Keep-alive

Sending messages

Error handling

See also

​Authentication

​Agent event stream

​Frame format

​Console timeline stream

​Identity timeline stream

​Event subscription

​WorkGraph wake stream

​Structural mob events stream

​Query parameters

​Frame format

​Stale cursor

​Continuation from mobkit/mob_events/subscribe

​Cursor durability

​Keep-alive

​Sending messages

​Error handling

​See also

Authentication

Agent event stream

Frame format

Console timeline stream

Identity timeline stream

Event subscription

WorkGraph wake stream

Structural mob events stream

Query parameters

Frame format

Stale cursor

Continuation from `mobkit/mob_events/subscribe`

Cursor durability

Keep-alive

Sending messages

Error handling

See also