mount_memory_status_route(router: Any, *, status: Any, resolve: Any) -> None
Mount status only when a deployment supplied its scoped service.
Source code in src/symfonic/platform/transport_memory_status.py
| def mount_memory_status_route(router: Any, *, status: Any, resolve: Any) -> None:
"""Mount status only when a deployment supplied its scoped service."""
if status is None:
return
from fastapi import Header, HTTPException, Request
@router.get("/memories/status")
async def memory_status(
request: Request,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> list[dict[str, Any]]:
scope = await resolve(x_tenant_id, authorization, request=request)
from symfonic.platform.wire import record_scope
# The Agent's declared identity when the router has one (kind-aware v1),
# else positional v0; a refused declaration stays typed, outside the catch-all.
memory = record_scope(scope)
try:
return await status.blocks(memory)
except Exception as unavailable: # noqa: BLE001
logger.error("Memory block status failed")
raise HTTPException(500, "Internal server error") from unavailable
|