FastAPI Request/Response Logging for AI Feature Attribution: Tracking Which Claude Call Cost Which Tenant
When your SaaS starts charging tenants for AI features, you enter a new debugging hell. A customer emails: "Why did my bill jump $340 last Tuesday?" You grep your logs. You find 50 Claude API calls that day. But which endpoint triggered them? Which user? Did they hit a bug that looped inference? You have no idea.
This is the problem I solved wrong three times before getting it right. Most tutorials show you how to log requests and responses separately. That's useless for AI attribution. You need correlated traces that follow a single request through your entire stack—from the HTTP endpoint, through your business logic, all the way to the Anthropic API call and its usage metrics.
I'm going to show you exactly how I implemented this in CitizenApp, and why every design choice matters when you're billing real customers.
The Core Problem: Request Isolation in Concurrent Systems
FastAPI handles hundreds of concurrent requests. When you log naively, you get this:
[2024-01-15 14:23:01] Request POST /generate-summary tenant_id=acme-corp
[2024-01-15 14:23:02] Claude API call usage: 450 input tokens, 120 output tokens
[2024-01-15 14:23:02] Request POST /generate-summary tenant_id=widgets-inc
[2024-01-15 14:23:03] Claude API response received
[2024-01-15 14:23:03] Claude API response received
Which response belongs to which tenant? You don't know. When you're processing $50k/month in inference spend across 200 tenants, this ambiguity is a nightmare.
The solution is a request-scoped context that persists across all async operations, paired with a structured logging format that includes the correlation ID in every single log line.
Implementation: Request Context and Structured Logging
I use Python's contextvars to store request metadata. It's thread-safe and async-safe, unlike thread-local storage. Here's the foundation:
# context.py
import contextvars
import uuid
from typing import Optional
from dataclasses import dataclass
@dataclass
class RequestContext:
request_id: str
tenant_id: str
user_id: str
endpoint: str
request_context: contextvars.ContextVar[Optional[RequestContext]] = contextvars.ContextVar(
'request_context',
default=None
)
def get_context() -> RequestContext:
ctx = request_context.get()
if not ctx:
raise RuntimeError("No request context set")
return ctx
def set_context(tenant_id: str, user_id: str, endpoint: str) -> str:
request_id = str(uuid.uuid4())
ctx = RequestContext(
request_id=request_id,
tenant_id=tenant_id,
user_id=user_id,
endpoint=endpoint
)
request_context.set(ctx)
return request_id
Now the middleware. This is where your request enters the system:
# middleware.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import logging
import time
from context import set_context, get_context
logger = logging.getLogger("attribution")
app = FastAPI()
@app.middleware("http")
async def add_request_context(request: Request, call_next):
# Extract tenant_id from JWT or header
auth_header = request.headers.get("Authorization", "")
token = auth_header.replace("Bearer ", "")
tenant_id = decode_tenant_from_jwt(token) # Your JWT logic
user_id = decode_user_from_jwt(token)
endpoint = request.url.path
request_id = set_context(tenant_id, user_id, endpoint)
start_time = time.time()
try:
response = await call_next(request)
duration = time.time() - start_time
logger.info("request_completed", extra={
"request_id": request_id,
"tenant_id": tenant_id,
"status_code": response.status_code,
"duration_ms": duration * 1000,
"endpoint": endpoint,
})
return response
except Exception as e:
duration = time.time() - start_time
logger.error("request_failed", extra={
"request_id": request_id,
"tenant_id": tenant_id,
"error": str(e),
"duration_ms": duration * 1000,
"endpoint": endpoint,
})
raise
Tracing Claude API Calls with Context
Now the critical part: when you call Anthropic, log the request and response with the same correlation ID:
# claude_service.py
import anthropic
import logging
from context import get_context
logger = logging.getLogger("attribution")
client = anthropic.Anthropic(api_key="your-key")
async def generate_summary(text: str) -> str:
ctx = get_context()
logger.info("claude_request_start", extra={
"request_id": ctx.request_id,
"tenant_id": ctx.tenant_id,
"model": "claude-3-5-sonnet-20241022",
"endpoint": ctx.endpoint,
})
try:
message = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
messages=[
{"role": "user", "content": f"Summarize: {text}"}
]
)
# Claude returns usage metrics—capture them immediately
usage = message.usage
logger.info("claude_response_success", extra={
"request_id": ctx.request_id,
"tenant_id": ctx.tenant_id,
"input_tokens": usage.input_tokens,
"output_tokens": usage.output_tokens,
"model": "claude-3-5-sonnet-20241022",
"endpoint": ctx.endpoint,
})
return message.content[0].text
except anthropic.APIError as e:
logger.error("claude_response_error", extra={
"request_id": ctx.request_id,
"tenant_id": ctx.tenant_id,
"error_code": e.status_code,
"error_message": str(e),
"endpoint": ctx.endpoint,
})
raise
Your endpoint looks normal, but now it's fully instrumented:
# routes.py
from fastapi import APIRouter, Depends
from claude_service import generate_summary
router = APIRouter()
@router.post("/generate-summary")
async def summarize_endpoint(payload: SummaryRequest):
# No logging code needed here—middleware and service handle it
result = await generate_summary(payload.text)
return {"summary": result}
Structured Logging Configuration
Use JSON logging so you can query it in your observability stack (DataDog, New Relic, etc.):
# logging_config.py
import logging
import json
from pythonjsonlogger import jsonlogger
def setup_logging():
logger = logging.getLogger("attribution")
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
# In your main.py
from logging_config import setup_logging
setup_logging()
Now every log line is JSON with consistent fields:
{
"message": "claude_response_success",
"request_id": "550e8400-e29b-41d4-a716-446655440000",
"tenant_id": "acme-corp",
"user_id": "user_42",
"input_tokens": 450,
"output_tokens": 120,
"endpoint": "/generate-summary",
"timestamp": "2024-01-15T14:23:02.123Z"
}
Query this in your logs:
tenant_id:"acme-corp" AND message:"claude_response_success"
| stats sum(input_tokens) as total_input, sum(output_tokens) as total_output by date
You instantly see: "On Jan 15, acme-corp used 45k input tokens and 12k output tokens across the summary feature."
Gotcha: Token Pricing Varies by Model
I initially lumped all tokens together in my cost calculation. Then I realized I was mixing claude-3-5-sonnet (cheaper) with claude-3-opus (expensive) in different features, and my attribution was off by 40%.
Fix: Log the model alongside usage, and store pricing tiers per-model in your database. Your billing query becomes:
# billing_service.py
PRICING = {
"claude-3-5-sonnet-20241022": {"input": 0.003, "output": 0.015},
"claude-3-opus-20250219": {"input": 0.015, "output": 0.075},
}
def calculate_tenant_costs(tenant_id: str, start_date: str, end_date: str):
logs = query_logs(f'tenant_id:"{tenant_id}" AND message:"claude_response_success"')
total_cost = 0
for log in logs:
model = log["model"]
input_tokens = log["input_tokens"]
output_tokens = log["output_tokens"]
cost = (
input_tokens * PRICING[model]["input"] +
output_tokens * PRICING[model]["output"]
) / 1000
total_cost += cost
return total_cost
Why This Matters
When a customer's bill spikes, you can now answer in seconds:
- Which endpoint triggered the spike? ✅
- Which user ran it? ✅
Top comments (0)