Observability Basics

9 examples to get you started with Observability - 6 basic and 3 intermediate.

Prerequisites

Install the official SDK: pip install anthropic.
Set ANTHROPIC_API_KEY in your environment.
Examples use only the standard library (logging, json, time, uuid) alongside anthropic, so no extra observability packages are required to follow along.

Basic Examples

1. Log a Single Call's Prompt and Response

The minimum viable observability: record what you sent and what came back.

import anthropic
 
client = anthropic.Anthropic()
 
prompt = "Summarize the risks of skipping code review."
 
response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": prompt}],
)
 
print(f"PROMPT: {prompt}")
print(f"RESPONSE: {response.content[0].text}")

response.content is a list of content blocks; [0].text gets the first text block.
This is the floor, not the ceiling - a print statement is fine for a one-off script but not for production.
The next examples turn this into a structured, queryable record instead of console output.

2. Capture Token Counts from the Response

Every response carries usage data; stop discarding it.

import anthropic
 
client = anthropic.Anthropic()
 
response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": "Explain idempotency in one paragraph."}],
)
 
usage = response.usage
print(f"input_tokens={usage.input_tokens} output_tokens={usage.output_tokens}")

response.usage exposes input_tokens and output_tokens on every non-streaming response.
Token counts are the raw material for cost tracking, so capture them even before you build a dashboard.
A common gotcha: token counts are per-response, not cumulative, so a multi-turn conversation needs you to sum them yourself.

Related: Structured Logging of Prompts, Responses, and Token Counts - the full schema this field feeds into

3. Write a Structured JSON Log Line

Replace ad hoc prints with one JSON object per call.

import json
import time
import anthropic
 
client = anthropic.Anthropic()
 
def call_and_log(prompt: str) -> str:
    start = time.monotonic()
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    )
    latency_ms = round((time.monotonic() - start) * 1000, 1)
 
    log_entry = {
        "model": response.model,
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
        "latency_ms": latency_ms,
        "prompt": prompt,
        "response": response.content[0].text,
    }
    print(json.dumps(log_entry))
    return response.content[0].text
 
call_and_log("List three benefits of feature flags.")

A JSON line per call is trivially parsed by any log aggregator (Datadog, CloudWatch, ELK) without a custom parser.
time.monotonic() avoids clock-adjustment issues that time.time() can introduce for latency measurement.
This is a single-function version of the pattern the structured logging article formalizes into a reusable schema.

4. Attach a Request ID for Correlation

A request ID lets you join a log line back to a trace span or a support ticket.

import uuid
import anthropic
 
client = anthropic.Anthropic()
 
def call_with_request_id(prompt: str) -> dict:
    request_id = str(uuid.uuid4())
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    )
    return {
        "request_id": request_id,
        "prompt": prompt,
        "response": response.content[0].text,
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
    }
 
result = call_with_request_id("Draft a one-line commit message for a bug fix.")
print(result["request_id"])

Generate the request ID before the call so it exists even if the call raises an exception.
Anthropic also returns its own _request_id on the response object; logging your own ID alongside it lets you correlate across your own systems as well as with Anthropic support if needed.
A common gotcha: forgetting to propagate the request ID into every downstream log line (tool calls, retries) breaks the correlation you built this for.

5. Log Errors with the Same Schema as Successes

A failed call still deserves a structured log entry, not just a stack trace.

import json
import anthropic
 
client = anthropic.Anthropic()
 
def safe_call(prompt: str) -> str | None:
    try:
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=300,
            messages=[{"role": "user", "content": prompt}],
        )
        print(json.dumps({"status": "ok", "prompt": prompt}))
        return response.content[0].text
    except anthropic.APIStatusError as exc:
        print(json.dumps({
            "status": "error",
            "prompt": prompt,
            "error_type": exc.__class__.__name__,
            "status_code": exc.status_code,
        }))
        return None
 
safe_call("Explain the CAP theorem briefly.")

anthropic.APIStatusError covers 4xx/5xx responses from the API; catch it explicitly instead of a bare except Exception.
Logging errors in the same JSON shape as successes means one dashboard query covers both, instead of hunting through separate error logs.
A common gotcha: swallowing the exception without logging status_code loses the distinction between a rate limit (429) and a server error (500), which need different responses.

6. Wrap the Client So Every Call Is Logged Automatically

Centralize logging instead of repeating it at every call site.

import json
import time
import anthropic
 
class ObservedClient:
    def __init__(self):
        self._client = anthropic.Anthropic()
 
    def create(self, **kwargs):
        start = time.monotonic()
        response = self._client.messages.create(**kwargs)
        latency_ms = round((time.monotonic() - start) * 1000, 1)
        print(json.dumps({
            "model": response.model,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "latency_ms": latency_ms,
        }))
        return response
 
client = ObservedClient()
client.create(
    model="claude-sonnet-5",
    max_tokens=200,
    messages=[{"role": "user", "content": "What is a bloom filter?"}],
)

Wrapping the SDK client means every call site gets logging for free, and you can't forget to add it to a new call.
This is the seed of the reusable pattern used throughout the rest of this section once you add tracing spans around create.
A common gotcha: wrapping only messages.create misses streaming calls (messages.stream), which need their own instrumentation.

Related: How Observability Works for LLM Applications - the mental model behind this wrapper pattern

Intermediate Examples

7. Log Every Turn in a Multi-Step Agent Loop

An agent loop makes several calls per user request; log each one with a shared identifier.

import json
import uuid
import anthropic
 
client = anthropic.Anthropic()
 
def run_agent_loop(user_prompt: str, max_steps: int = 3) -> str:
    run_id = str(uuid.uuid4())
    messages = [{"role": "user", "content": user_prompt}]
 
    for step in range(max_steps):
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=500,
            messages=messages,
        )
        print(json.dumps({
            "run_id": run_id,
            "step": step,
            "stop_reason": response.stop_reason,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
        }))
 
        if response.stop_reason != "tool_use":
            return response.content[0].text
 
        messages.append({"role": "assistant", "content": response.content})
        # A real agent would execute the tool call here and append its result.
        break
 
    return "Loop ended without a final answer."
 
run_agent_loop("What is 47 times 12? Show your work.")

The shared run_id is what lets you later group all steps of one agent run into a single trace-like view even before you add real OTel spans.
response.stop_reason tells you whether the model wants to call a tool (tool_use) or is done (end_turn), which is essential context for the log line.
This flat per-step logging is the precursor to the nested span structure built in the OpenTelemetry tracing article.

Related: Instrumenting Agent Loops with OpenTelemetry Tracing - turning these flat log steps into nested spans

8. Compute and Log an Estimated Cost Per Call

Token counts alone don't tell you spend; convert them with per-model pricing.

import json
import anthropic
 
client = anthropic.Anthropic()
 
# Illustrative rates only - verify current pricing at platform.claude.com/docs.
PRICE_PER_MILLION_TOKENS = {
    "claude-sonnet-5": {"input": 3.00, "output": 15.00},
    "claude-haiku-4.5": {"input": 0.80, "output": 4.00},
}
 
def call_and_log_cost(prompt: str, model: str = "claude-sonnet-5") -> str:
    response = client.messages.create(
        model=model,
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    )
    rates = PRICE_PER_MILLION_TOKENS[model]
    cost_usd = (
        response.usage.input_tokens * rates["input"]
        + response.usage.output_tokens * rates["output"]
    ) / 1_000_000
 
    print(json.dumps({"model": model, "cost_usd": round(cost_usd, 6)}))
    return response.content[0].text
 
call_and_log_cost("Give a one-sentence definition of eventual consistency.")

Pricing varies by model and changes over time, so keep the rate table in one place and treat it as configuration, not a hardcoded constant scattered across call sites.
Logging cost per call, not just tokens, is what makes a spend-spike alert possible later.
A common gotcha: forgetting cache-related token fields (cache_read_input_tokens) undercounts or overcounts cost once prompt caching is in use.

Related: Integrating Claude Usage and Cost Dashboards with Datadog - shipping this per-call cost into a dashboard

9. Log Streaming Responses Without Losing Token Counts

Streaming complicates logging because the response arrives in chunks.

import json
import anthropic
 
client = anthropic.Anthropic()
 
def stream_and_log(prompt: str) -> str:
    full_text = ""
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            full_text += text
        final_message = stream.get_final_message()
 
    print(json.dumps({
        "input_tokens": final_message.usage.input_tokens,
        "output_tokens": final_message.usage.output_tokens,
        "response_length": len(full_text),
    }))
    return full_text
 
stream_and_log("Write a two-sentence changelog entry for a bug fix release.")

stream.get_final_message() gives you the completed message, including final usage counts, after the stream finishes.
Logging only happens once the stream is fully consumed, so latency for a streamed call should be measured to the last chunk, not the first.
A common gotcha: logging inside the for text in stream.text_stream loop would fire once per chunk instead of once per call, flooding your logs.

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.

Observability Basics

9 examples to get you started with Observability - 6 basic and 3 intermediate.

Prerequisites

Install the official SDK: pip install anthropic.
Set ANTHROPIC_API_KEY in your environment.
Examples use only the standard library (logging, json, time, uuid) alongside anthropic, so no extra observability packages are required to follow along.

Basic Examples

1. Log a Single Call's Prompt and Response

The minimum viable observability: record what you sent and what came back.

import anthropic
 
client = anthropic.Anthropic()
 
prompt = "Summarize the risks of skipping code review."
 
response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": prompt}],
)
 
print(f"PROMPT: {prompt}")
print(f"RESPONSE: {response.content[0].text}")

response.content is a list of content blocks; [0].text gets the first text block.
This is the floor, not the ceiling - a print statement is fine for a one-off script but not for production.
The next examples turn this into a structured, queryable record instead of console output.

2. Capture Token Counts from the Response

Every response carries usage data; stop discarding it.

import anthropic
 
client = anthropic.Anthropic()
 
response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": "Explain idempotency in one paragraph."}],
)
 
usage = response.usage
print(f"input_tokens={usage.input_tokens} output_tokens={usage.output_tokens}")

response.usage exposes input_tokens and output_tokens on every non-streaming response.
Token counts are the raw material for cost tracking, so capture them even before you build a dashboard.
A common gotcha: token counts are per-response, not cumulative, so a multi-turn conversation needs you to sum them yourself.

Related: Structured Logging of Prompts, Responses, and Token Counts - the full schema this field feeds into

3. Write a Structured JSON Log Line

Replace ad hoc prints with one JSON object per call.

import json
import time
import anthropic
 
client = anthropic.Anthropic()
 
def call_and_log(prompt: str) -> str:
    start = time.monotonic()
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    )
    latency_ms = round((time.monotonic() - start) * 1000, 1)
 
    log_entry = {
        "model": response.model,
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
        "latency_ms": latency_ms,
        "prompt": prompt,
        "response": response.content[0].text,
    }
    print(json.dumps(log_entry))
    return response.content[0].text
 
call_and_log("List three benefits of feature flags.")

A JSON line per call is trivially parsed by any log aggregator (Datadog, CloudWatch, ELK) without a custom parser.
time.monotonic() avoids clock-adjustment issues that time.time() can introduce for latency measurement.
This is a single-function version of the pattern the structured logging article formalizes into a reusable schema.

4. Attach a Request ID for Correlation

A request ID lets you join a log line back to a trace span or a support ticket.

import uuid
import anthropic
 
client = anthropic.Anthropic()
 
def call_with_request_id(prompt: str) -> dict:
    request_id = str(uuid.uuid4())
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    )
    return {
        "request_id": request_id,
        "prompt": prompt,
        "response": response.content[0].text,
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
    }
 
result = call_with_request_id("Draft a one-line commit message for a bug fix.")
print(result["request_id"])

Generate the request ID before the call so it exists even if the call raises an exception.
Anthropic also returns its own _request_id on the response object; logging your own ID alongside it lets you correlate across your own systems as well as with Anthropic support if needed.
A common gotcha: forgetting to propagate the request ID into every downstream log line (tool calls, retries) breaks the correlation you built this for.

5. Log Errors with the Same Schema as Successes

A failed call still deserves a structured log entry, not just a stack trace.

import json
import anthropic
 
client = anthropic.Anthropic()
 
def safe_call(prompt: str) -> str | None:
    try:
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=300,
            messages=[{"role": "user", "content": prompt}],
        )
        print(json.dumps({"status": "ok", "prompt": prompt}))
        return response.content[0].text
    except anthropic.APIStatusError as exc:
        print(json.dumps({
            "status": "error",
            "prompt": prompt,
            "error_type": exc.__class__.__name__,
            "status_code": exc.status_code,
        }))
        return None
 
safe_call("Explain the CAP theorem briefly.")

anthropic.APIStatusError covers 4xx/5xx responses from the API; catch it explicitly instead of a bare except Exception.
Logging errors in the same JSON shape as successes means one dashboard query covers both, instead of hunting through separate error logs.
A common gotcha: swallowing the exception without logging status_code loses the distinction between a rate limit (429) and a server error (500), which need different responses.

6. Wrap the Client So Every Call Is Logged Automatically

Centralize logging instead of repeating it at every call site.

import json
import time
import anthropic
 
class ObservedClient:
    def __init__(self):
        self._client = anthropic.Anthropic()
 
    def create(self, **kwargs):
        start = time.monotonic()
        response = self._client.messages.create(**kwargs)
        latency_ms = round((time.monotonic() - start) * 1000, 1)
        print(json.dumps({
            "model": response.model,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "latency_ms": latency_ms,
        }))
        return response
 
client = ObservedClient()
client.create(
    model="claude-sonnet-5",
    max_tokens=200,
    messages=[{"role": "user", "content": "What is a bloom filter?"}],
)

Wrapping the SDK client means every call site gets logging for free, and you can't forget to add it to a new call.
This is the seed of the reusable pattern used throughout the rest of this section once you add tracing spans around create.
A common gotcha: wrapping only messages.create misses streaming calls (messages.stream), which need their own instrumentation.

Related: How Observability Works for LLM Applications - the mental model behind this wrapper pattern

Intermediate Examples

7. Log Every Turn in a Multi-Step Agent Loop

An agent loop makes several calls per user request; log each one with a shared identifier.

import json
import uuid
import anthropic
 
client = anthropic.Anthropic()
 
def run_agent_loop(user_prompt: str, max_steps: int = 3) -> str:
    run_id = str(uuid.uuid4())
    messages = [{"role": "user", "content": user_prompt}]
 
    for step in range(max_steps):
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=500,
            messages=messages,
        )
        print(json.dumps({
            "run_id": run_id,
            "step": step,
            "stop_reason": response.stop_reason,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
        }))
 
        if response.stop_reason != "tool_use":
            return response.content[0].text
 
        messages.append({"role": "assistant", "content": response.content})
        # A real agent would execute the tool call here and append its result.
        break
 
    return "Loop ended without a final answer."
 
run_agent_loop("What is 47 times 12? Show your work.")

The shared run_id is what lets you later group all steps of one agent run into a single trace-like view even before you add real OTel spans.
response.stop_reason tells you whether the model wants to call a tool (tool_use) or is done (end_turn), which is essential context for the log line.
This flat per-step logging is the precursor to the nested span structure built in the OpenTelemetry tracing article.

Related: Instrumenting Agent Loops with OpenTelemetry Tracing - turning these flat log steps into nested spans

8. Compute and Log an Estimated Cost Per Call

Token counts alone don't tell you spend; convert them with per-model pricing.

import json
import anthropic
 
client = anthropic.Anthropic()
 
# Illustrative rates only - verify current pricing at platform.claude.com/docs.
PRICE_PER_MILLION_TOKENS = {
    "claude-sonnet-5": {"input": 3.00, "output": 15.00},
    "claude-haiku-4.5": {"input": 0.80, "output": 4.00},
}
 
def call_and_log_cost(prompt: str, model: str = "claude-sonnet-5") -> str:
    response = client.messages.create(
        model=model,
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    )
    rates = PRICE_PER_MILLION_TOKENS[model]
    cost_usd = (
        response.usage.input_tokens * rates["input"]
        + response.usage.output_tokens * rates["output"]
    ) / 1_000_000
 
    print(json.dumps({"model": model, "cost_usd": round(cost_usd, 6)}))
    return response.content[0].text
 
call_and_log_cost("Give a one-sentence definition of eventual consistency.")

Pricing varies by model and changes over time, so keep the rate table in one place and treat it as configuration, not a hardcoded constant scattered across call sites.
Logging cost per call, not just tokens, is what makes a spend-spike alert possible later.
A common gotcha: forgetting cache-related token fields (cache_read_input_tokens) undercounts or overcounts cost once prompt caching is in use.

Related: Integrating Claude Usage and Cost Dashboards with Datadog - shipping this per-call cost into a dashboard

9. Log Streaming Responses Without Losing Token Counts

Streaming complicates logging because the response arrives in chunks.

import json
import anthropic
 
client = anthropic.Anthropic()
 
def stream_and_log(prompt: str) -> str:
    full_text = ""
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for text in stream.text_stream:
            full_text += text
        final_message = stream.get_final_message()
 
    print(json.dumps({
        "input_tokens": final_message.usage.input_tokens,
        "output_tokens": final_message.usage.output_tokens,
        "response_length": len(full_text),
    }))
    return full_text
 
stream_and_log("Write a two-sentence changelog entry for a bug fix release.")

stream.get_final_message() gives you the completed message, including final usage counts, after the stream finishes.
Logging only happens once the stream is fully consumed, so latency for a streamed call should be measured to the last chunk, not the first.
A common gotcha: logging inside the for text in stream.text_stream loop would fire once per chunk instead of once per call, flooding your logs.

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.

Observability Basics

Search Documentation

Observability Basics