def create_kernel_agent_router(
host: Any,
*,
prefix: str = DEFAULT_PREFIX,
memory: Any = None,
graph: Any = None,
scope_resolver: Any = None,
memory_audit: Any = None,
) -> Any:
"""A router that maps HTTP onto ``host``.
Args:
host: anything with ``async agent_for(scope)``. Structural, so a test
double or a decorating host stands in without inheriting.
prefix: where the routes mount.
memory_audit: explicit acknowledged audit sink for point deletion.
DELETE requires both this sink and a verified scope resolver;
a missing or failing audit sink prevents storage mutation.
scope_resolver: what turns a request's credentials into an
authenticated principal, normally a
:class:`~symfonic.platform.HeaderScopeResolver`. **Without it the
tenant header is taken on trust**, which is a development posture:
a caller who can set a header can then read any tenant. A
deployment that authenticates must pass one.
graph: an object with ``edges``, ``neighborhood`` and ``export`` --
normally a
:class:`~symfonic.capabilities.memory.graph_admin.GraphAdminService`.
Optional, and its absence is why the routes below are mounted
conditionally: a deployment with no graph view answers 404 rather
than returning an empty list a client would read as "no
relationships".
memory: an object with ``records(scope, layer=, limit=)`` -- normally a
:class:`~symfonic.capabilities.memory.admin.MemoryAdminService` over
the host's store. Optional: a deployment that exposes no memory
browser passes nothing and the route is not mounted at all, which
is a 404 rather than an endpoint that answers with an empty list
and lets a caller believe the tenant has no memories.
"""
try:
from fastapi import APIRouter, Body, Header, HTTPException
from fastapi.responses import StreamingResponse
except ImportError as missing: # pragma: no cover - exercised by the gate
# IA-4. A missing extra reaches the caller as an install command
# rather than as a bare import error: the person who hits this is
# standing up a deployment, and "no module named fastapi" does not
# tell them which extra of which distribution provides it.
raise ImportError(
"the kernel-native router needs FastAPI, which ships with the "
"'agent-api' extra: pip install 'symfonic-core[agent-api]'"
) from missing
router = APIRouter(prefix=prefix)
async def _authenticated_scope(
tenant: str | None, authorization: str | None, *, principal_only: bool = False,
) -> Any:
"""The scope this request is *allowed* to name, not the one it claims.
With a resolver, the tenant header is a claim the resolver checks
against the caller's credentials. Without one, the header is taken at
face value -- which is why the parameter exists and why a deployment
that skips it has no tenant isolation at the transport.
"""
if not tenant or not tenant.strip():
raise HTTPException(
status_code=401,
detail=f"missing {TENANT_HEADER}: the caller is unidentified, "
"so there is no tenant whose agent could serve this request",
)
if scope_resolver is None:
if principal_only:
raise HTTPException(401, "record deletion requires a verified identity")
return scope_for_tenant(tenant.strip())
principal = await verified_principal(scope_resolver, tenant.strip(), authorization)
return principal if principal_only else principal.scope
async def _agent_for(tenant: str | None, authorization: str | None) -> Any:
scope = await _authenticated_scope(tenant, authorization)
try:
return await host.agent_for(scope)
except HostClosed as closed:
raise HTTPException(
status_code=503,
detail="the host is shutting down and is composing no further "
"agents",
) from closed
def _query(payload: dict[str, Any]) -> str:
text = str(payload.get("query") or "").strip()
if not text:
raise HTTPException(
status_code=422,
detail="query must be a non-empty string",
)
return text
def _history(payload: dict[str, Any]) -> tuple[Any, ...]:
"""Validate the browser's transcript at the transport boundary.
``session_id`` is correlation, not conversation state. Treating it
as though the facade would load a transcript made every request from
the generated Chat page a first turn, even inside the same chat.
Only user/assistant text crosses this public endpoint: a caller may
not inject a system instruction or fabricate a tool result.
"""
raw = payload.get("history", ())
if raw in (None, ()):
return ()
if not isinstance(raw, list) or len(raw) > 100:
raise HTTPException(
status_code=422, detail="history must be a list of at most 100 messages"
)
from symfonic.agent.facade_types import Message
parsed = []
for item in raw:
if not isinstance(item, dict):
raise HTTPException(status_code=422, detail="invalid history message")
role = item.get("role")
content = item.get("content")
if role not in ("user", "assistant") or not isinstance(content, str):
raise HTTPException(
status_code=422,
detail="history messages require a user/assistant role and text content",
)
parsed.append(Message(role=role, content=content))
return tuple(parsed)
def _session_id(payload: dict[str, Any]) -> str:
value = payload.get("session_id", "")
if not isinstance(value, str):
raise HTTPException(status_code=422, detail="session_id must be text")
value = value.strip()
if len(value) > 64:
raise HTTPException(status_code=422, detail="session_id too long (max 64)")
return value
@router.post("/chat")
async def chat(
payload: dict[str, Any] = Body(...), # noqa: B008 - FastAPI idiom
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> dict[str, Any]:
agent = await _agent_for(x_tenant_id, authorization)
try:
result = await agent.run(
_query(payload), history=_history(payload), session_id=_session_id(payload)
)
except HTTPException:
raise
except asyncio.CancelledError:
# Never mapped: a cancelled request has no response to send, and
# swallowing it here would leave the turn's teardown to a task
# nobody is waiting on.
raise
except Exception as failed: # noqa: BLE001
# A turn that fails upstream is a 502, not a 500: the caller's
# request was well formed and the fault is behind us. Left
# unmapped it escaped as an unhandled ASGI exception, which some
# clients re-raise instead of reporting a status at all.
raise HTTPException(
status_code=502,
detail=f"the turn failed: {failed}",
) from failed
return {"response": text_of(result)}
@router.post("/stream")
async def stream(
payload: dict[str, Any] = Body(...), # noqa: B008 - FastAPI idiom
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> Any:
agent = await _agent_for(x_tenant_id, authorization)
query = _query(payload)
history = _history(payload)
async def events():
async for chunk in agent.stream(
query, history=history, session_id=_session_id(payload)
):
# Only chunks that CARRY text. A stream yields terminal events
# too -- the last one holds the whole result and no text -- and
# falling back to ``str(event)`` for those put a full repr on
# the wire as though it were content. A client would render it.
text = getattr(chunk, "text", None)
if isinstance(text, str) and text:
yield f"data: {json.dumps({'chunk': text})}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(events(), media_type="text/event-stream")
@router.post("/stream/typed")
async def stream_typed(
payload: dict[str, Any] = Body(...), # noqa: B008 - FastAPI idiom
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> Any:
"""The same turn, with each event's kind on the wire.
Derived from ``stream`` rather than from a typed API, because the
facade has one stream and it already carries the kind. Every event is
forwarded -- including the ones with no text, which is the difference
from ``/stream``: a client that needs to tell a tool call from a token
needs the events ``/stream`` deliberately drops.
"""
agent = await _agent_for(x_tenant_id, authorization)
query = _query(payload)
history = _history(payload)
async def events():
async for event in agent.stream(
query, history=history, session_id=_session_id(payload)
):
body: dict[str, Any] = {"type": kind_of(event)}
for field in ("text", "error", "index"):
value = getattr(event, field, None)
if value is not None:
body[field] = value
yield f"data: {json.dumps(body)}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(events(), media_type="text/event-stream")
mount_data_routes(
router, memory=memory, graph=graph, resolve=_authenticated_scope
)
mount_record_routes(router, memory=memory, resolve=_authenticated_scope,
principal=_authenticated_scope, sink=memory_audit)
return router