Skip to content

symfonic.agent.fastapi.metrics_router

metrics_router

Metrics dashboard endpoints for token usage and cost observability.

create_metrics_router 

create_metrics_router(
    metrics_collector: ConversationMetricsCollector,
    prefix: str = "/api/v1",
) -> APIRouter

Create a FastAPI router exposing conversation metrics endpoints.

Endpoints

GET {prefix}/metrics/conversations?limit=50 GET {prefix}/metrics/conversations/{conversation_id} GET {prefix}/metrics/summary?start=&end=

Parameters:

Name Type Description Default
metrics_collector ConversationMetricsCollector

The ConversationMetricsCollector to expose.

 required
prefix str

URL prefix for the router (default /api/v1).

 '/api/v1'

Returns:

Type Description
APIRouter

Configured APIRouter.
Source code in src/symfonic/agent/fastapi/metrics_router.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def create_metrics_router(
    metrics_collector: ConversationMetricsCollector,
    prefix: str = "/api/v1",
) -> APIRouter:
    """Create a FastAPI router exposing conversation metrics endpoints.

    Endpoints:
        GET {prefix}/metrics/conversations?limit=50
        GET {prefix}/metrics/conversations/{conversation_id}
        GET {prefix}/metrics/summary?start=&end=

    Args:
        metrics_collector: The ConversationMetricsCollector to expose.
        prefix: URL prefix for the router (default ``/api/v1``).

    Returns:
        Configured APIRouter.
    """
    router = APIRouter(prefix=prefix, tags=["metrics"])

    @router.get("/metrics/conversations")
    async def list_conversations(
        limit: int = 50,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> list[dict[str, Any]]:
        """Return aggregated metrics for up to ``limit`` conversations.

        Sorted by last_seen descending (most recently active first).
        Restricted to the caller's tenant.
        """
        records = metrics_collector.list_conversations(
            limit=limit, tenant_id=str(scope.tenant_id)
        )
        return [
            {
                "conversation_id": r.conversation_id,
                "tenant_id": r.tenant_id,
                "turns": r.turns,
                "llm_calls": r.llm_calls,
                "input_tokens": r.input_tokens,
                "output_tokens": r.output_tokens,
                "cached_tokens": r.cached_tokens,
                "cost_usd": r.cost_usd,
                "first_seen": r.first_seen.isoformat(),
                "last_seen": r.last_seen.isoformat(),
            }
            for r in records
        ]

    @router.get("/metrics/conversations/{conversation_id}")
    async def get_conversation(
        conversation_id: str,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> dict[str, Any]:
        """Return the full ConversationRecord including per-call breakdown.

        Returns 404 (never 403) when the conversation belongs to a different
        tenant so existence is not leaked across tenant boundaries.
        """
        record = metrics_collector.get_conversation(conversation_id)
        if record is None or record.tenant_id != str(scope.tenant_id):
            raise HTTPException(
                status_code=404,
                detail=f"Conversation {conversation_id!r} not found",
            )
        return {
            "conversation_id": record.conversation_id,
            "tenant_id": record.tenant_id,
            "turns": record.turns,
            "llm_calls": record.llm_calls,
            "input_tokens": record.input_tokens,
            "output_tokens": record.output_tokens,
            "cached_tokens": record.cached_tokens,
            "cost_usd": record.cost_usd,
            "first_seen": record.first_seen.isoformat(),
            "last_seen": record.last_seen.isoformat(),
            "call_records": [
                {
                    "call_id": cr.call_id,
                    "model": cr.model,
                    "node": cr.node,
                    "input_tokens": cr.input_tokens,
                    "cached_tokens": cr.cached_tokens,
                    "output_tokens": cr.output_tokens,
                    "cost_usd": cr.cost_usd,
                    "call_type": cr.call_type,
                    "timestamp": cr.timestamp.isoformat(),
                    "composition": cr.composition,
                }
                for cr in record.call_records
            ],
        }

    @router.get("/metrics/summary")
    async def get_summary(
        start: datetime | None = None,
        end: datetime | None = None,
        scope: FrameworkTenantScope = Depends(get_tenant_scope),  # noqa: B008
    ) -> dict[str, Any]:
        """Return aggregate totals across the caller's tenant.

        Optionally filtered by ``start`` and/or ``end`` datetime (ISO-8601).
        """
        return metrics_collector.summary(
            start=start, end=end, tenant_id=str(scope.tenant_id)
        )

    return router