def mount_data_routes(
router: Any, *, memory: Any, graph: Any, resolve: Any
) -> None:
"""Add the store-backed routes to ``router``.
Args:
router: the ``APIRouter`` the agent routes are already on.
memory: a ``MemoryAdminService``, or ``None``.
graph: a ``GraphAdminService``, or ``None``.
resolve: the transport's authenticated-scope resolver. Passed in
rather than rebuilt, so these routes cannot end up authorising
differently from the ones next to them.
"""
from fastapi import Header, HTTPException, Response
if memory is not None:
@router.get("/memories")
async def list_memories(
response: Response,
layer: str | None = None,
limit: int = 200,
cursor: str | None = None,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> list[dict[str, Any]]:
"""A published inventory page in stable record-ID order.
Reads through the memory admin service, so it sees exactly what a
turn would recall -- the same store, the same scope, the same
published-only rule. A browser built on a second query path is a
browser that can disagree with the agent.
"""
scope = await resolve(x_tenant_id, authorization)
try:
page = await memory.inventory_page(
memory_scope(scope), layer=memory_layer(layer), limit=limit, cursor=cursor,
)
except NotImplementedError as unsupported:
raise HTTPException(status_code=501, detail=str(unsupported)) from unsupported
except ValueError as unknown:
raise HTTPException(status_code=400, detail=str(unknown)) from unknown
response.headers["X-Memory-Has-More"] = str(page.next_cursor is not None).lower()
if page.next_cursor is not None:
response.headers["X-Memory-Next-Cursor"] = page.next_cursor
response.headers["Access-Control-Expose-Headers"] = (
"X-Memory-Has-More, X-Memory-Next-Cursor"
)
return [record_body(record) for record in page.records]
if graph is not None:
@router.get("/graph/nodes")
async def list_graph_nodes(
limit: int = 200,
offset: int = 0,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> list[dict[str, Any]]:
"""The nodes paired with the graph relationship endpoint."""
scope = await resolve(x_tenant_id, authorization)
if limit < 1 or limit > 500 or offset < 0:
raise HTTPException(
status_code=400,
detail="limit must be 1..500 and offset must be non-negative",
)
return await graph.nodes(
memory_scope(scope), limit=limit, offset=offset
)
@router.get("/edges")
async def list_edges(
limit: int = 200,
offset: int = 0,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> list[dict[str, Any]]:
"""The relationships this tenant can see."""
scope = await resolve(x_tenant_id, authorization)
if limit < 1 or limit > 500 or offset < 0:
raise HTTPException(
status_code=400,
detail="limit must be 1..500 and offset must be non-negative",
)
return await graph.edges(
memory_scope(scope), limit=limit, offset=offset
)
@router.get("/graph/nodes/{node_id}/neighborhood")
async def neighborhood(
node_id: str,
depth: int = 1,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> dict[str, Any]:
"""One node's neighbours, out to ``depth`` hops."""
scope = await resolve(x_tenant_id, authorization)
return await graph.neighborhood(memory_scope(scope), node_id, depth=depth)
@router.get("/tenants/me/export")
async def export_tenant_data(
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> dict[str, Any]:
"""GDPR Article 20 -- everything this tenant owns, as one document.
A partial export still answers the request and says so in
``complete``: a portability response that silently omitted a layer
would be a failure nobody could see.
"""
scope = await resolve(x_tenant_id, authorization)
return await graph.export(memory_scope(scope))
if memory is not None:
@router.delete("/tenants/me/data")
async def erase_tenant_data(
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
) -> dict[str, Any]:
"""GDPR Article 17 -- erase this tenant and everything beneath it.
``forget`` walks the subtree, which is the direction erasure has to
run: a request that deleted a tenant and left its sessions behind
would report success and leave the data.
"""
scope = await resolve(x_tenant_id, authorization)
erased = memory_scope(scope)
receipt = await memory.forget(erased)
return {
"scope_path": erased.path,
"erased": getattr(receipt, "count", None),
"receipt": str(receipt),
}