def create_kernel_agent_router(
host: Any,
*,
prefix: str = DEFAULT_PREFIX,
memory: Any = None,
memory_updates: Any = None,
semantic_create: Any = None,
episodic_create: Any = None,
procedural_create: Any = None,
prospective_create: Any = None,
memory_status: Any = None,
graph: Any = None, edge_delete: Any = None,
privacy: Any = None, procedures: Any = None, admin_target_authorizer: Any = None,
scope_resolver: Any = None,
memory_audit: Any = None, graph_audit: Any = None,
conversation: Any = None,
working_access: Any = None,
working_create: Any = None, memory_identity: 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.
privacy: opt-in PrivacyService for verified export and confirmed erasure.
admin_target_authorizer: verified caller and target scope to bool grant.
memory_audit: acknowledged DELETE sink; failure blocks mutation.
memory_identity: host-composed scope for record read/delete.
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**, 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`.
Absent graph means no graph routes (404).
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, Request
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
from symfonic.platform.transport_production_gate import gate_native_http_assembly
gate_native_http_assembly(scope_resolver) # before any route, agent, model or tool
from symfonic.platform.transport_sse_response import NativeStreamResponse
graph, graph_audit = owner_read_port(graph), owner_read_port(graph_audit)
from symfonic.platform.transport_scope_refusal import refusing_route_class as routes
router = APIRouter(prefix=prefix, route_class=routes(memory_identity))
async def _authenticated_scope(
tenant: str | None,
authorization: str | None,
*,
request: Request | None = None,
principal_only: bool = False,
with_principal: 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; without a resolver this is development-only.
"""
gate_native_http_assembly(scope_resolver)
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, "deletion requires a verified identity")
scope = scope_for_tenant(tenant.strip())
return (scope, None) if with_principal else scope
principal = await verified_principal(
scope_resolver, tenant.strip(), authorization, request=request,
)
if with_principal:
return principal.scope, principal
return principal if principal_only else principal.scope
async def working(tenant, authorization, *, request):
from symfonic.platform.working_access import NativeWorkingAccess
scope, caller = await _authenticated_scope(
tenant, authorization, request=request, with_principal=True,
)
return scope, NativeWorkingAccess(caller, working_access)
if working_create is not None:
from symfonic.platform.transport_working_create import mount_working_create_route
mount_working_create_route(router, host=host, service=working_create,
resolve=_authenticated_scope, verified_resolver=scope_resolver is not None)
async def _agent_for(
tenant: str | None, authorization: str | None, payload: dict[str, Any], request: Request,
) -> tuple[Any, str, tuple[Any, ...], str, Any]:
if conversation is not None:
principal = await _authenticated_scope(
tenant, authorization, request=request, principal_only=True,
)
from symfonic.platform.working_access import verified_owner_principal
if not verified_owner_principal(principal):
raise HTTPException(401, "chat requires a verified principal")
scope = principal.scope
else:
principal = None
scope = await _authenticated_scope(tenant, authorization, request=request)
query, session_id = query_of(payload), session_id_of(payload)
refuse_attachments(payload)
if session_id and conversation is None:
raise HTTPException(503, "session owner admission is not configured")
if conversation is not None:
if not session_id:
raise HTTPException(422, "session_id is required for this chat route")
# The deployment verifies this exact principal owns the session.
# Browser history is untrusted whenever the server holds that chat.
prepared = await conversation.prepare(principal, session_id, query)
history = prepared.history
else:
history, prepared = history_of(payload), None
try:
agent = await host.agent_for(scope)
require_supported_agent(agent)
except UnsupportedAgentComposer as unsupported:
raise HTTPException(status_code=500, detail=str(unsupported)) from unsupported
except HostClosed as closed:
detail = "the host is shutting down and is composing no further agents"
raise HTTPException(status_code=503, detail=detail) from closed
except Exception as failed: # CORE-235: a refused FrameworkConfig, before egress
if refused_setting(failed):
raise HTTPException(status_code=502, detail="the turn failed") from failed
raise
return agent, query, history, session_id, prepared
@router.post("/chat")
async def chat(
request: Request,
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, query, history, session_id, prepared = await _agent_for(
x_tenant_id, authorization, payload, request
)
try:
result = await agent.run(query, history=history, session_id=session_id)
answer = visible_answer(answer_of_run(agent, result))
if prepared is not None:
await prepared.publish(answer)
except UnusableFinal as unusable:
raise HTTPException(502, "turn ended without a usable answer") from unusable
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,
# Exception text can include private configuration or provider data.
detail="the turn failed",
) from failed
return {"response": answer}
@router.post("/stream")
async def stream(
request: Request,
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, query, history, session_id, prepared = await _agent_for(
x_tenant_id, authorization, payload, request
)
return NativeStreamResponse(
agent, query, history, session_id, prepared, typed=False,
)
@router.post("/stream/typed")
async def stream_typed(
request: Request,
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.
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, query, history, session_id, prepared = await _agent_for(
x_tenant_id, authorization, payload, request
)
return NativeStreamResponse(
agent, query, history, session_id, prepared, typed=True,
)
from symfonic.platform.transport_memory_status import mount_memory_status_route
mount_memory_status_route(router, status=memory_status, resolve=_authenticated_scope)
mount_data_routes(router, memory=memory, graph=graph, edge_delete=edge_delete,
resolve=_authenticated_scope, principal=_authenticated_scope,
memory_audit=memory_audit, privacy=privacy,
procedures=procedures, scope_resolver=scope_resolver, working=working)
if graph_audit is not None:
from symfonic.platform.transport_graph_audit import mount_graph_audit_route
mount_graph_audit_route(router, audit=graph_audit, resolve=_authenticated_scope,
header=Header, http_exception=HTTPException,
request_type=Request, working=working)
if privacy is not None:
from symfonic.platform.transport_privacy import mount_privacy_routes
mount_privacy_routes(
router, privacy=privacy, resolver=scope_resolver,
admin_target_authorizer=admin_target_authorizer)
mount_record_routes(
router,
memory=memory,
updates=memory_updates,
resolve=_authenticated_scope,
principal=_authenticated_scope,
sink=memory_audit,
verified_resolver=scope_resolver is not None,
working=working,
)
from symfonic.platform.transport_semantic_create import mount_semantic_create_route
mount_semantic_create_route(
router, service=semantic_create, resolve=_authenticated_scope,
verified_resolver=scope_resolver is not None)
from symfonic.platform.transport_episodic_create import mount_episodic_create_route
mount_episodic_create_route(
router, service=episodic_create, resolve=_authenticated_scope,
verified_resolver=scope_resolver is not None)
from symfonic.platform.transport_prospective_create import mount_prospective_create_route
mount_prospective_create_route(
router, service=prospective_create, resolve=_authenticated_scope,
verified_resolver=scope_resolver is not None)
mount_continuation_route(router, host=host, resolve=_authenticated_scope)
from symfonic.platform.transport_procedural_create import mount_procedural_create_route
mount_procedural_create_route(
router, host=host, service=procedural_create, resolve=_authenticated_scope,
verified_resolver=scope_resolver is not None)
return router