Muscles SSE

Server-Sent Events adapter for Muscles streaming use cases.

This package should provide progress, logs, notifications, and AI output streams over SSE while keeping the underlying workflow in the Muscles application model.

Related Repositories

• muscles - core StreamResult, StreamEvent, actions and dispatcher.
• muscles-asgi - HTTP runtime that can expose SSE responses.
• muscles-ai - AI/RAG actions that may produce streaming output.
• muscles-jsonrpc - sibling protocol projection over actions.
• muscles-benchmarks - streaming regression checks.

Concept Guardrails

• SSE streams must be backed by Muscles actions/events/jobs, not ad hoc handlers.
• The adapter must preserve shared rules/security and request context.
• Stream events must have typed names and payload schemas.
• SSE should be the simple default for one-way streaming before WebSocket is introduced.
• The package must be usable from ASGI without forcing unrelated runtime choices.

Implemented Features

Implemented SSE transport projection over Muscles action execution:

• action dispatch path: dispatcher.execute(action, payload, transport="sse");
• application facade: SseAdapter.from_application(app) backed by core ActionDispatcher;
• stream normalization delegates to core stream_events() / StreamResult;
• typed events: progress, log, result, error;
• wire formatting: id, event, retry, data;
• SseResponse with SSE headers and status;
• configurable interval heartbeat with bounded backpressure;
• safe source close on stream completion/disconnect;
• structured core action error mapping: not found, validation, permission, execution.

This keeps SSE as a thin delivery layer. Business logic stays in Muscles actions.

SseAdapter.execute() returns an SseResponse, which carries the event stream and SSE response metadata. It is a protocol result, not a framework HTTP response: the hosting ASGI integration must forward its status and headers and stream its body to the client.

Interval heartbeat

SSE serializes core StreamEvent items from StreamResult. Business events come from muscles.core; this package owns the SSE wire format and transport heartbeat, backpressure, and disconnect behavior.

adapter = SseAdapter(
    dispatcher,
    heartbeat_event="heartbeat",
    heartbeat_interval_seconds=15,
)

When the action stream is quiet longer than the configured interval, the transport emits a heartbeat event:

event: heartbeat
data: {"ok": true}

User events keep their existing SSE format. Closing the response stream stops the heartbeat worker and closes the underlying source when it supports close(). Backpressure is preserved with a bounded one-item queue. Long- blocking sources should implement a cooperative close() that unblocks the active next() call.

Run tests

python -m pytest -q