Noções Básicas de Observabilidade

9 exemplos para você começar com Observabilidade - 6 básicos e 3 intermediários.

Pré-requisitos

Instale o SDK oficial: pip install anthropic.
Defina ANTHROPIC_API_KEY em seu ambiente.
Os exemplos usam apenas a biblioteca padrão (logging, json, time, uuid) junto com anthropic, portanto, nenhum pacote de observabilidade extra é necessário para acompanhar.

Exemplos Básicos

1. Registrar o Prompt e a Resposta de Uma Única Chamada

A observabilidade mínima viável: registre o que você enviou e o que retornou.

import anthropic
 
client = anthropic.Anthropic()
 
prompt = "Resuma os riscos de pular a revisão de código."
 
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 é uma lista de blocos de conteúdo; [0].text pega o primeiro bloco de texto.
Este é o mínimo, não o máximo - uma instrução print é suficiente para um script único, mas não para produção.
Os próximos exemplos transformam isso em um registro estruturado e consultável em vez de saída de console.

2. Capturar Contagens de Tokens da Resposta

Toda resposta carrega dados de uso; pare de descartá-los.

import anthropic
 
client = anthropic.Anthropic()
 
response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": "Explique idempotência em um parágrafo."}],
)
 
usage = response.usage
print(f"input_tokens={usage.input_tokens} output_tokens={usage.output_tokens}")

response.usage expõe input_tokens e output_tokens em cada resposta não-streaming.
As contagens de tokens são a matéria-prima para o rastreamento de custos, portanto, capture-as mesmo antes de construir um painel.
Um problema comum: as contagens de tokens são por resposta, não cumulativas, então uma conversa com vários turnos exige que você as some manualmente.

Relacionado: Registro Estruturado de Prompts, Respostas e Contagens de Tokens - o esquema completo ao qual este campo alimenta

3. Escrever Uma Linha de Log JSON Estruturada

Substitua impressões ad hoc por um objeto JSON por chamada.

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("Liste três benefícios de flags de recursos.")

Uma linha JSON por chamada é trivialmente analisada por qualquer agregador de logs (Datadog, CloudWatch, ELK) sem um analisador personalizado.
time.monotonic() evita problemas de ajuste de relógio que time.time() pode introduzir para medição de latência.
Esta é uma versão de função única do padrão que o artigo de registro estruturado formaliza em um esquema reutilizável.

4. Anexar um ID de Requisição para Correlação

Um ID de requisição permite que você vincule uma linha de log a um span de rastreamento ou a um ticket de suporte.

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("Esboce uma mensagem de commit de uma linha para uma correção de bug.")
print(result["request_id"])

Gere o ID da requisição antes da chamada para que ele exista mesmo que a chamada gere uma exceção.
A Anthropic também retorna seu próprio _request_id no objeto de resposta; registrar seu próprio ID ao lado dele permite correlacionar entre seus próprios sistemas, bem como com o suporte da Anthropic, se necessário.
Um problema comum: esquecer de propagar o ID da requisição para todas as linhas de log downstream (chamadas de ferramentas, novas tentativas) quebra a correlação para a qual você o construiu.

5. Registrar Erros com o Mesmo Esquema que Sucessos

Uma chamada falha ainda merece uma entrada de log estruturada, não apenas um 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("Explique o teorema CAP brevemente.")

anthropic.APIStatusError cobre respostas 4xx/5xx da API; capture-a explicitamente em vez de um except Exception genérico.
Registrar erros na mesma forma JSON que sucessos significa que uma consulta de painel cobre ambos, em vez de procurar em logs de erro separados.
Um problema comum: engolir a exceção sem registrar status_code perde a distinção entre um limite de taxa (429) e um erro de servidor (500), que exigem respostas diferentes.

6. Envolver o Cliente para que Toda Chamada Seja Registrada Automaticamente

Centralize o registro em vez de repeti-lo em cada local de chamada.

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": "O que é um filtro de bloom?"}],
)

Envolver o cliente do SDK significa que cada local de chamada obtém o registro gratuitamente, e você não pode esquecer de adicioná-lo a uma nova chamada.
Esta é a semente do padrão reutilizável usado em toda a seção restante, uma vez que você adiciona spans de rastreamento em torno de create.
Um problema comum: envolver apenas messages.create perde chamadas de streaming (messages.stream), que precisam de sua própria instrumentação.

Relacionado: Como a Observabilidade Funciona para Aplicações LLM - o modelo mental por trás deste padrão de wrapper

Exemplos Intermediários

7. Registrar Cada Turno em um Loop de Agente Multi-Etapas

Um loop de agente faz várias chamadas por solicitação do usuário; registre cada uma com um identificador compartilhado.

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})
        # Um agente real executaria a chamada da ferramenta aqui e adicionaria seu resultado.
        break
 
    return "Loop encerrado sem uma resposta final."
 
run_agent_loop("Qual é 47 vezes 12? Mostre seu trabalho.")

O run_id compartilhado é o que permite agrupar posteriormente todas as etapas de uma execução de agente em uma única visualização semelhante a um rastreamento, mesmo antes de adicionar spans OTel reais.
response.stop_reason informa se o modelo deseja chamar uma ferramenta (tool_use) ou se terminou (end_turn), o que é um contexto essencial para a linha de log.
Este registro plano por etapa é o precursor da estrutura de span aninhada construída no artigo de rastreamento OpenTelemetry.

Relacionado: Instrumentando Loops de Agente com Rastreamento OpenTelemetry - transformando essas etapas de log planas em spans aninhados

8. Calcular e Registrar um Custo Estimado por Chamada

Apenas as contagens de tokens não informam o gasto; converta-as com preços por modelo.

import json
import anthropic
 
client = anthropic.Anthropic()
 
# Taxas ilustrativas apenas - verifique os preços atuais em 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("Dê uma definição de uma frase de consistência eventual.")

Os preços variam por modelo e mudam ao longo do tempo, portanto, mantenha a tabela de taxas em um só lugar e trate-a como configuração, não como uma constante codificada em vários locais de chamada.
Registrar o custo por chamada, não apenas os tokens, é o que torna um alerta de pico de gastos possível mais tarde.
Um problema comum: esquecer os campos de token relacionados ao cache (cache_read_input_tokens) subconta ou superconta o custo assim que o cache de prompt estiver em uso.

Relacionado: Integrando Dashboards de Uso e Custo do Claude com Datadog - enviando este custo por chamada para um dashboard

9. Registrar Respostas de Streaming Sem Perder Contagens de Tokens

O streaming complica o registro porque a resposta chega em blocos.

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("Escreva uma entrada de changelog de duas frases para um lançamento de correção de bug.")

stream.get_final_message() fornece a mensagem completa, incluindo as contagens finais de usage, após o término do stream.
O registro só ocorre quando o stream é totalmente consumido, portanto, a latência para uma chamada transmitida deve ser medida até o último bloco, não o primeiro.
Um problema comum: registrar dentro do loop for text in stream.text_stream dispararia uma vez por bloco em vez de uma vez por chamada, inundando seus logs.

Versões da Stack: Escrito contra a linha de modelos Claude atual em aproximadamente junho de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (o padrão), e Claude Haiku 4.5 - e o SDK oficial anthropic para Python (última versão 0.x). Nomes de modelos, preços e versões de SDK mudam rapidamente - verifique os detalhes atuais em platform.claude.com/docs antes de confiar neles.

Noções Básicas de Observabilidade

9 exemplos para você começar com Observabilidade - 6 básicos e 3 intermediários.

Pré-requisitos

Instale o SDK oficial: pip install anthropic.
Defina ANTHROPIC_API_KEY em seu ambiente.
Os exemplos usam apenas a biblioteca padrão (logging, json, time, uuid) junto com anthropic, portanto, nenhum pacote de observabilidade extra é necessário para acompanhar.

Exemplos Básicos

1. Registrar o Prompt e a Resposta de Uma Única Chamada

A observabilidade mínima viável: registre o que você enviou e o que retornou.

import anthropic
 
client = anthropic.Anthropic()
 
prompt = "Resuma os riscos de pular a revisão de código."
 
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 é uma lista de blocos de conteúdo; [0].text pega o primeiro bloco de texto.
Este é o mínimo, não o máximo - uma instrução print é suficiente para um script único, mas não para produção.
Os próximos exemplos transformam isso em um registro estruturado e consultável em vez de saída de console.

2. Capturar Contagens de Tokens da Resposta

Toda resposta carrega dados de uso; pare de descartá-los.

import anthropic
 
client = anthropic.Anthropic()
 
response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": "Explique idempotência em um parágrafo."}],
)
 
usage = response.usage
print(f"input_tokens={usage.input_tokens} output_tokens={usage.output_tokens}")

response.usage expõe input_tokens e output_tokens em cada resposta não-streaming.
As contagens de tokens são a matéria-prima para o rastreamento de custos, portanto, capture-as mesmo antes de construir um painel.
Um problema comum: as contagens de tokens são por resposta, não cumulativas, então uma conversa com vários turnos exige que você as some manualmente.

Relacionado: Registro Estruturado de Prompts, Respostas e Contagens de Tokens - o esquema completo ao qual este campo alimenta

3. Escrever Uma Linha de Log JSON Estruturada

Substitua impressões ad hoc por um objeto JSON por chamada.

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("Liste três benefícios de flags de recursos.")

Uma linha JSON por chamada é trivialmente analisada por qualquer agregador de logs (Datadog, CloudWatch, ELK) sem um analisador personalizado.
time.monotonic() evita problemas de ajuste de relógio que time.time() pode introduzir para medição de latência.
Esta é uma versão de função única do padrão que o artigo de registro estruturado formaliza em um esquema reutilizável.

4. Anexar um ID de Requisição para Correlação

Um ID de requisição permite que você vincule uma linha de log a um span de rastreamento ou a um ticket de suporte.

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("Esboce uma mensagem de commit de uma linha para uma correção de bug.")
print(result["request_id"])

Gere o ID da requisição antes da chamada para que ele exista mesmo que a chamada gere uma exceção.
A Anthropic também retorna seu próprio _request_id no objeto de resposta; registrar seu próprio ID ao lado dele permite correlacionar entre seus próprios sistemas, bem como com o suporte da Anthropic, se necessário.
Um problema comum: esquecer de propagar o ID da requisição para todas as linhas de log downstream (chamadas de ferramentas, novas tentativas) quebra a correlação para a qual você o construiu.

5. Registrar Erros com o Mesmo Esquema que Sucessos

Uma chamada falha ainda merece uma entrada de log estruturada, não apenas um 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("Explique o teorema CAP brevemente.")

anthropic.APIStatusError cobre respostas 4xx/5xx da API; capture-a explicitamente em vez de um except Exception genérico.
Registrar erros na mesma forma JSON que sucessos significa que uma consulta de painel cobre ambos, em vez de procurar em logs de erro separados.
Um problema comum: engolir a exceção sem registrar status_code perde a distinção entre um limite de taxa (429) e um erro de servidor (500), que exigem respostas diferentes.

6. Envolver o Cliente para que Toda Chamada Seja Registrada Automaticamente

Centralize o registro em vez de repeti-lo em cada local de chamada.

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": "O que é um filtro de bloom?"}],
)

Envolver o cliente do SDK significa que cada local de chamada obtém o registro gratuitamente, e você não pode esquecer de adicioná-lo a uma nova chamada.
Esta é a semente do padrão reutilizável usado em toda a seção restante, uma vez que você adiciona spans de rastreamento em torno de create.
Um problema comum: envolver apenas messages.create perde chamadas de streaming (messages.stream), que precisam de sua própria instrumentação.

Relacionado: Como a Observabilidade Funciona para Aplicações LLM - o modelo mental por trás deste padrão de wrapper

Exemplos Intermediários

7. Registrar Cada Turno em um Loop de Agente Multi-Etapas

Um loop de agente faz várias chamadas por solicitação do usuário; registre cada uma com um identificador compartilhado.

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})
        # Um agente real executaria a chamada da ferramenta aqui e adicionaria seu resultado.
        break
 
    return "Loop encerrado sem uma resposta final."
 
run_agent_loop("Qual é 47 vezes 12? Mostre seu trabalho.")

O run_id compartilhado é o que permite agrupar posteriormente todas as etapas de uma execução de agente em uma única visualização semelhante a um rastreamento, mesmo antes de adicionar spans OTel reais.
response.stop_reason informa se o modelo deseja chamar uma ferramenta (tool_use) ou se terminou (end_turn), o que é um contexto essencial para a linha de log.
Este registro plano por etapa é o precursor da estrutura de span aninhada construída no artigo de rastreamento OpenTelemetry.

Relacionado: Instrumentando Loops de Agente com Rastreamento OpenTelemetry - transformando essas etapas de log planas em spans aninhados

8. Calcular e Registrar um Custo Estimado por Chamada

Apenas as contagens de tokens não informam o gasto; converta-as com preços por modelo.

import json
import anthropic
 
client = anthropic.Anthropic()
 
# Taxas ilustrativas apenas - verifique os preços atuais em 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("Dê uma definição de uma frase de consistência eventual.")

Os preços variam por modelo e mudam ao longo do tempo, portanto, mantenha a tabela de taxas em um só lugar e trate-a como configuração, não como uma constante codificada em vários locais de chamada.
Registrar o custo por chamada, não apenas os tokens, é o que torna um alerta de pico de gastos possível mais tarde.
Um problema comum: esquecer os campos de token relacionados ao cache (cache_read_input_tokens) subconta ou superconta o custo assim que o cache de prompt estiver em uso.

Relacionado: Integrando Dashboards de Uso e Custo do Claude com Datadog - enviando este custo por chamada para um dashboard

9. Registrar Respostas de Streaming Sem Perder Contagens de Tokens

O streaming complica o registro porque a resposta chega em blocos.

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("Escreva uma entrada de changelog de duas frases para um lançamento de correção de bug.")

stream.get_final_message() fornece a mensagem completa, incluindo as contagens finais de usage, após o término do stream.
O registro só ocorre quando o stream é totalmente consumido, portanto, a latência para uma chamada transmitida deve ser medida até o último bloco, não o primeiro.
Um problema comum: registrar dentro do loop for text in stream.text_stream dispararia uma vez por bloco em vez de uma vez por chamada, inundando seus logs.

Versões da Stack: Escrito contra a linha de modelos Claude atual em aproximadamente junho de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (o padrão), e Claude Haiku 4.5 - e o SDK oficial anthropic para Python (última versão 0.x). Nomes de modelos, preços e versões de SDK mudam rapidamente - verifique os detalhes atuais em platform.claude.com/docs antes de confiar neles.

Noções Básicas de Observabilidade

Buscar na documentação

Noções Básicas de Observabilidade