Skip to content

symfonic.agent.fastapi.native_typed_stream

native_typed_stream

Opt-in AgentHost compatibility /stream/typed route.

Only stateless Agent turns enter this route. Raw event admission and usage proof precede the typed projection, which owns the frozen class-name SSE wire.

create_native_typed_stream_router 

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

Mount the native typed route separately from the live legacy router.

Source code in src/symfonic/agent/fastapi/native_typed_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
 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
def create_native_typed_stream_router(
    host: Any,
    *,
    sessions: Any,
    prefix: str = "/api/v1",
    app: Any = None,
) -> APIRouter:
    """Mount the native typed route separately from the live legacy router."""
    if not callable(getattr(host, "agent_for", None)):
        raise TypeError("native typed stream host must provide agent_for(scope)")
    if not all(callable(getattr(sessions, name, None)) for name in ("ensure_session", "touch")):
        raise TypeError("native typed stream sessions must provide ensure_session and touch")
    _check_production_auth_gate(app)
    router = APIRouter(prefix=prefix, tags=["agent"])

    @router.post("/stream/typed")
    async def stream_typed(
        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 = _visible_events(typed_stream(_raw_events(
                agent, invocation, session_id,
                admission.metered, 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, encode_typed_event))

    return router