Skip to content

symfonic.agent.fastapi.shared_stream

shared_stream

Selected-root /stream over the shared native conversation authority.

The frames are the frozen compatibility chunk contract (STR-2): one SSE frame per StreamChunk, a done terminal carrying final_response, and the stream policy's error frame. Ownership, attachment family, capability and budget metering are refused before the model is called. A turn is published to the shared transcript only when its final answer is usable, and before the done frame is sent. No SymfonicAgent is constructed or called here.

Session issuance: a first turn (no session_id) gets a session from the authority, and its id is returned in the X-Session-ID response header. It is not added to any frame. The id is bound to the verified principal, tenant, sub-tenant and namespace, so a continuation from any other identity is refused like any unowned id. The session row exists from issuance, before the first frame, but the first turn is persisted only once the done frame is sent. A session whose first turn publishes nothing (an error frame, a client disconnect, a response that was never iterated, or a prepare failure before the response) is discarded; a failed discard is logged, not raised. An unknown client-chosen session_id is still refused; nothing is adopted.

Declared divergence from the legacy root: a turn that completes with an empty answer ends in the stream policy's error frame and publishes nothing, where legacy emits done with final_response=''. A tool-limit stop is refused the same way; its legacy frame sequence is not compared here. Both follow the selected /chat, which refuses the same unusable finals (answer_of_result). The frame shapes are shared with legacy; full wire parity is not claimed.

create_shared_stream_router 

create_shared_stream_router(host: Any, conversation: Any) -> APIRouter

Mount /stream for the shared root; the caller owns prefix and app.

Source code in src/symfonic/agent/fastapi/shared_stream.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def create_shared_stream_router(host: Any, conversation: Any) -> APIRouter:
    """Mount ``/stream`` for the shared root; the caller owns prefix and app."""
    router = APIRouter()

    @router.post("/stream")
    async def stream(
        request: AgentRequest,
        raw_request: Request,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> Any:
        turn, agent, admission = await open_shared_turn(
            host, conversation, request, raw_request, scope
        )
        return TurnResponse(
            turn,
            SseFrameAdapter(STREAM_POLICY).frames(
                _chunks(turn.events(agent, admission)),
                lambda chunk: {"event": chunk.event_type, "data": chunk.model_dump_json()},
            ),
            headers=turn.headers,
        )

    return router