symfonic.agent.fastapi.native_chat

native_chat

Opt-in AgentHost /chat root with the compatibility response envelope.

This is a first consumer cutover, not a replacement for create_agent_router: the chat router mounts only /chat, requires an Agent with no behavior capabilities, and keeps the compatibility request, auth, budget, error and response boundaries. A separate opt-in router lists those native sessions. Deployments needing memory, streaming, continuation or the other legacy endpoints keep their existing router until those contracts are migrated.

create_native_chat_router

create_native_chat_router(host: Any, *, sessions: Any, conversation: Any = None, prefix: str = '/api/v1', app: Any = None, _owner_claimed: bool = False) -> APIRouter

Mount one stateless Agent-backed chat route using AgentResponse JSON.

host.agent_for(scope) must return a public Agent with no behavior capabilities. The built-in telemetry capability is admitted only when its folded sink owner proves the authenticated tenant and standard collector; budgeted requests without that proof are refused before model execution. sessions is an explicit scope-aware identity port with ensure_session and touch (the public SessionManager satisfies it). The port is shared by requests; making it inside a handler would silently lose cross-tenant collision protection on every new request.

The composition root opts in by mounting this router instead of the compatibility router for /chat. Conversation-backed composition requires the outermost FastAPI app; mounting that selected app beneath another router is unsupported. No SymfonicAgent is constructed or called here; no switch is flipped for any other legacy endpoint.

Source code in src/symfonic/agent/fastapi/native_chat.py

def create_native_chat_router(
    host: Any,
    *,
    sessions: Any,
    conversation: Any = None,
    prefix: str = "/api/v1",
    app: Any = None,
    _owner_claimed: bool = False,
) -> APIRouter:
    """Mount one stateless Agent-backed chat route using AgentResponse JSON.

    ``host.agent_for(scope)`` must return a public ``Agent`` with no behavior
    capabilities. The built-in telemetry capability is admitted only when its
    folded sink owner proves the authenticated tenant and standard collector;
    budgeted requests without that proof are refused before model execution.
    ``sessions`` is an explicit scope-aware identity port with
    ``ensure_session`` and ``touch`` (the public ``SessionManager`` satisfies
    it). The port is shared by requests; making it inside a handler would
    silently lose cross-tenant collision protection on every new request.

    The composition root opts in by mounting this router instead of the
    compatibility router for /chat. Conversation-backed composition requires
    the outermost FastAPI app; mounting that selected app beneath another
    router is unsupported. No SymfonicAgent is constructed or called
    here; no switch is flipped for any other legacy endpoint.
    """
    if not callable(getattr(host, "agent_for", None)):
        raise TypeError("native chat host must provide agent_for(scope)")
    if conversation is not None:
        if app is None:
            raise ValueError("native conversation chat owner requires an explicit FastAPI app")
        from symfonic.agent.fastapi.conversation_http import validate_authority

        validate_authority(conversation)
    elif not all(callable(getattr(sessions, name, None)) for name in ("ensure_session", "touch")):
        raise TypeError("native chat sessions must provide ensure_session and touch")
    _check_production_auth_gate(app)
    router = APIRouter(prefix=prefix, tags=["agent"])

    @router.post("/chat", response_model=AgentResponse)
    async def chat(
        request: AgentRequest,
        raw_request: Request,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> AgentResponse:
        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),
        )
        issued = None
        published = False
        try:
            prepared = None
            if conversation is not None:
                from symfonic.agent.fastapi.conversation_http import principal_of

                principal = principal_of(raw_request, scope)
                if invocation.attachments:
                    from symfonic.platform.conversation_authority import ConversationFamilyRefused

                    raise ConversationFamilyRefused(
                        "shared local conversation history admits text-only chat"
                    )
                if invocation.session_id:
                    session_id = invocation.session_id
                    prepared = await conversation.prepare(principal, session_id, invocation.query)
            agent = await host.agent_for(scope)
            if not isinstance(agent, Agent) or agent.capabilities not in ((), ("telemetry",)):
                raise ConfigurationError(
                    "native compatibility chat requires a stateless Agent "
                    "without behavior capabilities"
                )
            if agent.capabilities or get_budget_tracker() is not None:
                from symfonic.platform.telemetry import budget_metering_is_bound

                if not budget_metering_is_bound(agent, str(scope.tenant_id)):
                    raise SymfonicAgentError(
                        "native chat cannot attest tenant-bound budget usage",
                        code="native_chat_unavailable",
                    )
            if conversation is None:
                session_id = sessions.ensure_session(str(scope.tenant_id), invocation.session_id)
                sessions.touch(session_id)
            elif prepared is None:
                issued = session_id = await conversation.create(principal)
                prepared = await conversation.prepare(principal, session_id, invocation.query)
            with bind_admin_authority(invocation.admin_claim()):
                result = await agent.run(
                    invocation.query,
                    attachments=tuple(invocation.attachments or ()),
                    session_id=session_id,
                    **({"history": prepared.history} if prepared is not None else {}),
                )
            if prepared is not None:
                from symfonic.platform.transport_conversation import answer_of_result

                await prepared.publish(visible_answer(answer_of_result(result)))
                published = True
            from symfonic.agent.response_projection import as_response

            projected = as_response(result, run_id=result.run_id, session_id=session_id)
            return projected.model_copy(update={
                "final_response": (
                    visible_answer(projected.final_response)
                    if projected.final_response else None
                ),
                "messages": visible_assistant_messages(projected.messages),
            })
        except Exception as exc:
            from symfonic.platform.host import HostClosed

            if isinstance(exc, HostClosed):
                raise raise_mapped(
                    SymfonicAgentError(
                        "the host is shutting down and is composing no further agents",
                        code="native_chat_unavailable",
                    ),
                    CHAT_POLICY,
                ) from exc
            raise raise_mapped(exc, CHAT_POLICY) from exc
        finally:
            if issued is not None and not published:
                try:
                    await conversation.discard_empty(principal, issued)
                except Exception:
                    logging.getLogger(__name__).warning(
                        "could not discard empty conversation after unsuccessful first turn",
                        exc_info=True,
                    )

    if not _owner_claimed:
        claim_http_chat_owner(app, prefix, "native")
    return router

create_native_chats_router

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

List a tenant's sessions beside opt-in native chat or stream routes.

Source code in src/symfonic/agent/fastapi/native_chat.py

def create_native_chats_router(
    sessions: Any,
    *,
    prefix: str = "/api/v1",
    app: Any = None,
) -> APIRouter:
    """List a tenant's sessions beside opt-in native chat or stream routes."""
    if not callable(getattr(sessions, "list_sessions", None)):
        raise TypeError("native chats sessions must provide list_sessions(tenant_id)")
    _check_production_auth_gate(app)
    router = APIRouter(prefix=prefix, tags=["agent"])

    @router.get("/chats")
    async def list_chats(
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> list[dict[str, Any]]:
        try:
            return sessions.list_sessions(str(scope.tenant_id))
        except Exception as exc:
            raise raise_mapped(exc, CRUD_POLICY) from exc

    return router