Skip to content

symfonic.agent.fastapi.native_stream

native_stream

Opt-in AgentHost-backed compatibility /stream endpoint.

Only the stateless chunk projection is admitted. With a budget tracker, both the tenant-bound telemetry sink and a declared provider streaming-usage contract are required before model work; every model round must then publish observed usage before a successful SSE terminal. The legacy router remains the entry point for typed streams, continuation, and capability-bearing turns.

create_native_stream_router 

create_native_stream_router(host: Any, *, sessions: Any, prefix: str = '/api/v1', app: Any = None) -> APIRouter

Mount /stream separately from the still-live compatibility router.

Source code in src/symfonic/agent/fastapi/native_stream.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def create_native_stream_router(
    host: Any,
    *,
    sessions: Any,
    prefix: str = "/api/v1",
    app: Any = None,
) -> APIRouter:
    """Mount ``/stream`` separately from the still-live compatibility router."""
    if not callable(getattr(host, "agent_for", None)):
        raise TypeError("native stream host must provide agent_for(scope)")
    if not all(callable(getattr(sessions, name, None)) for name in ("ensure_session", "touch")):
        raise TypeError("native stream sessions must provide ensure_session and touch")
    _check_production_auth_gate(app)
    router = APIRouter(prefix=prefix, tags=["agent"])

    @router.post("/stream")
    async def stream(
        request: AgentRequest,
        raw_request: Request,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> Any:
        from sse_starlette.sse import EventSourceResponse

        await check_budget_for_tenant(raw_request, str(scope.tenant_id))
        invocation = _CODEC.decode_invocation(
            request,
            scope=scope,
            is_admin=request_is_admin(raw_request),
            principal_id=request_user_id(raw_request),
        )
        try:
            agent = await host.agent_for(scope)
            admission = admit_native_stream(agent, str(scope.tenant_id), invocation.query)
            session_id = sessions.ensure_session(str(scope.tenant_id), invocation.session_id)
            sessions.touch(session_id)
            source = _chunks(
                agent, invocation, session_id,
                metered=admission.metered,
                prepared_plan=admission.prepared_plan,
            )
        except Exception as exc:
            from symfonic.platform.host import HostClosed

            if isinstance(exc, HostClosed):
                exc = SymfonicAgentError(
                    "the host is shutting down and is composing no further agents",
                    code="native_stream_unavailable",
                )
            raise raise_mapped(exc, STREAM_POLICY) from exc

        adapter = SseFrameAdapter(STREAM_POLICY)
        return EventSourceResponse(
            adapter.frames(
                source,
                lambda chunk: {"event": chunk.event_type, "data": chunk.model_dump_json()},
            )
        )

    return router