Structured Logging of Prompts, Responses, and Token Counts

A structured log is a consistent, machine-parseable record of one Claude API call: the prompt, the response, the token counts, the model, the latency, and a request ID.

Without that consistency, every call site logs something slightly different, and answering "how many tokens did we use yesterday" turns into grepping through inconsistent text instead of running a query.

Summary

The goal of structured logging is a fixed schema applied to every call, not a best-effort log message.

Each field in the schema earns its place: the prompt and response for debugging quality issues, token counts for cost, latency for performance, model name for comparing across model versions, and a request ID for correlating with traces or support tickets.

A Python logging.Logger with a JSON formatter, or a dedicated logging adapter function, is enough to enforce this without adopting a new framework.

The pattern generalizes past a single messages.create call: streaming responses, multi-turn conversations, and tool-using agent loops all need the same schema applied per call, not per conversation.

This page builds a reusable logging adapter, then extends it to multi-turn and streaming cases.

Recipe

import json
import logging
import time
import uuid
 
logger = logging.getLogger("claude.calls")
 
def log_call(*, request_id, model, prompt, response_text, usage, latency_ms, status="ok"):
    logger.info(json.dumps({
        "request_id": request_id,
        "model": model,
        "prompt": prompt,
        "response": response_text,
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "latency_ms": latency_ms,
        "status": status,
    }))

When to reach for this:

• You have more than one call site making Claude API calls and want a consistent, queryable log format.
• You need to answer "how many tokens did this feature use last week" from logs, not by re-running requests.
• You are about to add tracing or cost dashboards, both of which depend on this schema existing first.
• You want error and success logs in the same shape so one query covers both.

Working Example

import json
import logging
import time
import uuid
 
import anthropic
 
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("claude.calls")
 
client = anthropic.Anthropic()
 
 
def log_entry(**fields) -> None:
    """Emit one structured JSON log line, dropping any field left as None."""
    logger.info(json.dumps({k: v for k, v in fields.items() if v is not None}))
 
 
def call_claude(prompt: str, model: str = "claude-sonnet-5") -> str:
    """Make one Claude call and log a structured entry for it, success or failure."""
    request_id = str(uuid.uuid4())
    start = time.monotonic()
 
    try:
        response = client.messages.create(
            model=model,
            max_tokens=500,
            messages=[{"role": "user", "content": prompt}],
        )
    except anthropic.APIStatusError as exc:
        log_entry(
            request_id=request_id,
            model=model,
            prompt=prompt,
            response=None,
            input_tokens=None,
            output_tokens=None,
            latency_ms=round((time.monotonic() - start) * 1000, 1),
            status="error",
            error_type=exc.__class__.__name__,
            status_code=exc.status_code,
        )
        raise
 
    response_text = response.content[0].text
    log_entry(
        request_id=request_id,
        model=response.model,
        prompt=prompt,
        response=response_text,
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        cache_read_input_tokens=getattr(response.usage, "cache_read_input_tokens", None),
        latency_ms=round((time.monotonic() - start) * 1000, 1),
        status="ok",
    )
    return response_text
 
 
def call_claude_multi_turn(messages: list[dict], model: str = "claude-sonnet-5") -> dict:
    """Same schema, applied to one turn of a multi-turn conversation."""
    request_id = str(uuid.uuid4())
    start = time.monotonic()
 
    response = client.messages.create(
        model=model,
        max_tokens=500,
        messages=messages,
    )
 
    log_entry(
        request_id=request_id,
        model=response.model,
        prompt=messages[-1]["content"],
        response=response.content[0].text if response.content else "",
        input_tokens=response.usage.input_tokens,
        output_tokens=response.usage.output_tokens,
        latency_ms=round((time.monotonic() - start) * 1000, 1),
        status="ok",
        turn_count=len(messages),
    )
    return {"request_id": request_id, "response": response}
 
 
if __name__ == "__main__":
    call_claude("Summarize the tradeoffs of microservices in two sentences.")

What this demonstrates:

• A single log_entry function enforcing one schema, called from both the success and error paths.
• Dropping None fields keeps error log lines from carrying misleading empty response/token fields.
• request_id generated before the call, so it exists even when the call raises.
• cache_read_input_tokens captured with getattr since it's only present once prompt caching applies.
• The same schema reused for a multi-turn call, with an added turn_count field, showing the schema extends rather than gets replaced per call type.

Deep Dive

How It Works

• The schema is just a Python dict serialized to JSON; there's no special library required, though a dedicated JSON logging library (python-json-logger, structlog) removes the manual json.dumps boilerplate at scale.
• Centralizing the log call in one function (log_entry) is what actually enforces the schema; if every call site builds its own dict inline, fields drift apart over time.
• logging.Logger (not print) is used because it integrates with log level filtering, handlers, and most log aggregation agents (Datadog Agent, Fluent Bit, CloudWatch Logs agent) out of the box.
• Error and success paths use the same field set with status distinguishing them, so a single query like status:error or an aggregation over input_tokens works regardless of outcome.

Schema Reference

Python Notes

• Prefer logging.Logger over print even for scripts; it costs nothing extra and means the same code works unmodified once you add handlers for a real log aggregator.
• Use logger.info(json.dumps(...)) for a minimal setup; for higher-throughput services, a logging.Formatter subclass that JSON-encodes the whole LogRecord avoids double-serialization and keeps stack traces intact on errors.
• Redact or truncate prompt/response fields before logging if your application handles regulated or sensitive user data; a structured schema makes this easier since redaction becomes one function, not a per-call-site decision.

Gotchas

• Building the log dict inline at every call site. Without a shared log_entry function, one call site adds a field and another doesn't, and your schema silently drifts. Fix: centralize log construction in one function or a small logging adapter class.
• Logging usage fields before checking for an exception. If the call raises, response never gets assigned, and referencing response.usage in a finally block crashes with NameError. Fix: compute token fields only inside the success branch, and log None for them on the error path.
• Missing cache_read_input_tokens breaking cost math downstream. Cache-related usage fields aren't always present on the response object depending on SDK version and whether caching applied. Fix: access them with getattr(response.usage, "cache_read_input_tokens", None) rather than direct attribute access.
• Logging full prompt/response text at a very high log level in production. This can balloon log storage costs and, if prompts contain user PII, creates a compliance surface you didn't intend. Fix: apply redaction/truncation before logging, and confirm your log retention policy for this stream matches your data handling requirements.
• Reusing print(json.dumps(...)) instead of the logging module. This bypasses log level filtering and most log aggregation agents, which expect log lines on stdout/stderr via a logging framework or expect a specific file format. Fix: always route through logging.Logger, even for a quick script.

Alternatives

FAQs

try:
    response = client.messages.create(...)
except anthropic.APIStatusError as exc:
    # response was never assigned; log None for usage fields instead
    log_entry(status="error", error_type=exc.__class__.__name__, input_tokens=None)
    raise

Related

• Observability Basics - the smaller, single-function version of this pattern.
• How Observability Works for LLM Applications - where structured logging fits relative to tracing and alerting.
• Instrumenting Agent Loops with OpenTelemetry Tracing - joining these log lines to trace spans via request ID.
• Integrating Claude Usage and Cost Dashboards with Datadog - shipping this schema's token fields into a cost dashboard.
• Observability Best Practices - a consolidated checklist across this section.

Stack versions: Written against the Claude model lineup current as of ~June 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (the default), and Claude Haiku 4.5 - and the official anthropic Python SDK (latest 0.x release). Model names, pricing, and SDK versions move quickly - verify current specifics at platform.claude.com/docs before relying on them.