def create_metrics_router(
collector: Any = None, *, scope_resolver: Any = None, prefix: str = "/api/v1",
read_guard: Any = None,
) -> Any:
"""Read the existing collector with the native transport's auth posture.
With a resolver, credentials determine the tenant; without one, the tenant
header is trusted for development, just like ``create_kernel_agent_router``.
Production assembly and reads fail closed unless that posture authenticates.
No collector mounts no routes. Storage and aggregation policy stay with the
supplied collector, including its historical limit and date semantics.
"""
try:
from fastapi import APIRouter, Depends, Header, HTTPException, Request
except ImportError as missing:
raise ImportError(
"the native metrics router needs FastAPI: "
"pip install 'symfonic-core[agent-api]'"
) from missing
from symfonic.platform.transport_production_gate import gate_native_http_assembly
from symfonic.platform.transport_records import verified_principal
if collector is not None:
gate_native_http_assembly(scope_resolver)
router = APIRouter(prefix=prefix, tags=["metrics"])
if collector is None:
return router
async def tenant_for(tenant: str | None, authorization: str | None, request: Request) -> str:
gate_native_http_assembly(scope_resolver)
if not tenant or not tenant.strip():
raise HTTPException(401, "missing X-Tenant-ID")
if scope_resolver is None:
return tenant.strip()
principal = await verified_principal(
scope_resolver, tenant.strip(), authorization, request=request,
)
return str(principal.scope.tenant_id)
@asynccontextmanager
async def unguarded(tenant: str):
yield
async def tenant_access(
request: Request,
x_tenant_id: str | None = Header(default=None),
authorization: str | None = Header(default=None),
):
tenant = await tenant_for(x_tenant_id, authorization, request)
# A host may hold its durable erasure barrier through response creation.
async with (read_guard or unguarded)(tenant):
yield tenant
@router.get("/metrics/conversations")
async def list_conversations(
limit: int = 50,
tenant: str = Depends(tenant_access), # noqa: B008
) -> list[dict[str, Any]]:
return [_conversation_body(record) for record in collector.list_conversations(
limit=limit, tenant_id=tenant,
)]
@router.get("/metrics/conversations/{conversation_id}")
async def get_conversation(
conversation_id: str,
tenant: str = Depends(tenant_access), # noqa: B008
) -> dict[str, Any]:
record = collector.get_conversation(conversation_id, tenant_id=tenant)
if record is None or record.tenant_id != tenant:
raise HTTPException(404, f"Conversation {conversation_id!r} not found")
return {
**_conversation_body(record),
"call_records": [{
"call_id": call.call_id,
"model": call.model,
"node": call.node,
"input_tokens": call.input_tokens,
"cached_tokens": call.cached_tokens,
"output_tokens": call.output_tokens,
"cost_usd": call.cost_usd,
"call_type": call.call_type,
"timestamp": call.timestamp.isoformat(),
"composition": call.composition,
} for call in record.call_records],
}
@router.get("/metrics/summary")
async def get_summary(
start: datetime | None = None,
end: datetime | None = None,
tenant: str = Depends(tenant_access), # noqa: B008
) -> dict[str, Any]:
return collector.summary(start=start, end=end, tenant_id=tenant)
return router