Conceptos básicos de observabilidad

9 ejemplos para empezar con la observabilidad: 6 básicos y 3 intermedios.

Prerrequisitos

Instala el SDK oficial: pip install anthropic.
Configura ANTHROPIC_API_KEY en tu entorno.
Los ejemplos solo usan la biblioteca estándar (logging, json, time, uuid) junto con anthropic, por lo que no se requieren paquetes de observabilidad adicionales para seguir el ejemplo.

Ejemplos básicos

1. Registrar el prompt y la respuesta de una sola llamada

La observabilidad mínima viable: registra lo que enviaste y lo que recibiste.

import anthropic
 
client = anthropic.Anthropic()
 
prompt = "Resume los riesgos de omitir la revisión 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"RESPUESTA: {response.content[0].text}")

response.content es una lista de bloques de contenido; [0].text obtiene el primer bloque de texto.
Este es el mínimo, no el máximo: una instrucción print está bien para un script único, pero no para producción.
Los siguientes ejemplos convierten esto en un registro estructurado y consultable en lugar de una salida de consola.

2. Capturar recuentos de tokens de la respuesta

Cada respuesta contiene datos de uso; deja de descartarlos.

import anthropic
 
client = anthropic.Anthropic()
 
response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": "Explica la idempotencia en un párrafo."}],
)
 
usage = response.usage
print(f"input_tokens={usage.input_tokens} output_tokens={usage.output_tokens}")

response.usage expone input_tokens y output_tokens en cada respuesta no streaming.
Los recuentos de tokens son la materia prima para el seguimiento de costos, así que captúralos incluso antes de crear un panel.
Una dificultad común: los recuentos de tokens son por respuesta, no acumulativos, por lo que una conversación de varios turnos requiere que los sumes tú mismo.

Relacionado: Registro estructurado de prompts, respuestas y recuentos de tokens - el esquema completo al que alimenta este campo.

3. Escribir una línea de registro JSON estructurada

Reemplaza las impresiones ad hoc con un objeto JSON por llamada.

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("Enumera tres beneficios de las feature flags.")

Una línea JSON por llamada es analizada trivialmente por cualquier agregador de registros (Datadog, CloudWatch, ELK) sin un analizador personalizado.
time.monotonic() evita problemas de ajuste del reloj que time.time() puede introducir para la medición de latencia.
Esta es una versión de una sola función del patrón que el artículo de registro estructurado formaliza en un esquema reutilizable.

4. Adjuntar un ID de solicitud para la correlación

Un ID de solicitud te permite vincular una línea de registro a un span de rastreo o a un ticket de soporte.

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("Redacta un mensaje de commit de una línea para una corrección de errores.")
print(result["request_id"])

Genera el ID de solicitud antes de la llamada para que exista incluso si la llamada genera una excepción.
Anthropic también devuelve su propio _request_id en el objeto de respuesta; registrar tu propio ID junto con él te permite correlacionar entre tus propios sistemas, así como con el soporte de Anthropic si es necesario.
Una dificultad común: olvidar propagar el ID de solicitud a cada línea de registro posterior (llamadas a herramientas, reintentos) rompe la correlación para la que lo creaste.

5. Registrar errores con el mismo esquema que los éxitos

Una llamada fallida todavía merece una entrada de registro estructurada, no solo un 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("Explica brevemente el teorema CAP.")

anthropic.APIStatusError cubre las respuestas 4xx/5xx de la API; captúrala explícitamente en lugar de un except Exception genérico.
Registrar errores con la misma forma JSON que los éxitos significa que una consulta de panel cubre ambos, en lugar de buscar en registros de errores separados.
Una dificultad común: tragar la excepción sin registrar status_code pierde la distinción entre un límite de tasa (429) y un error del servidor (500), que requieren respuestas diferentes.

6. Envolver el cliente para que cada llamada se registre automáticamente

Centraliza el registro en lugar de repetirlo en cada sitio de llamada.

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": "¿Qué es un filtro bloom?"}],
)

Envolver el cliente SDK significa que cada sitio de llamada obtiene el registro de forma gratuita, y no puedes olvidar agregarlo a una nueva llamada.
Esta es la semilla del patrón reutilizable utilizado en el resto de esta sección una vez que agregues spans de rastreo alrededor de create.
Una dificultad común: envolver solo messages.create omite las llamadas de streaming (messages.stream), que necesitan su propia instrumentación.

Relacionado: Cómo funciona la observabilidad para aplicaciones LLM - el modelo mental detrás de este patrón de envoltorio.

Ejemplos intermedios

7. Registrar cada turno en un bucle de agente de varios pasos

Un bucle de agente realiza varias llamadas por solicitud de usuario; registra cada una con un identificador compartido.

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})
        # Un agente real ejecutaría la llamada a la herramienta aquí y agregaría su resultado.
        break
 
    return "El bucle terminó sin una respuesta final."
 
run_agent_loop("¿Cuánto es 47 por 12? Muestra tu trabajo.")

El run_id compartido es lo que te permite agrupar posteriormente todos los pasos de una ejecución de agente en una única vista similar a un rastreo, incluso antes de agregar spans reales de OTel.
response.stop_reason te dice si el modelo quiere llamar a una herramienta (tool_use) o ha terminado (end_turn), lo cual es contexto esencial para la línea de registro.
Este registro plano por paso es el precursor de la estructura de spans anidados construida en el artículo de rastreo de OpenTelemetry.

Relacionado: Instrumentación de bucles de agente con rastreo de OpenTelemetry - convertir estos pasos de registro planos en spans anidados.

8. Calcular y registrar un costo estimado por llamada

Los recuentos de tokens por sí solos no te dicen el gasto; conviértelos con precios por modelo.

import json
import anthropic
 
client = anthropic.Anthropic()
 
# Tarifas ilustrativas solamente: verifica los precios actuales en 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("Da una definición de una oración de consistencia eventual.")

Los precios varían según el modelo y cambian con el tiempo, así que mantén la tabla de tarifas en un solo lugar y trátala como configuración, no como una constante codificada en los sitios de llamada.
Registrar el costo por llamada, no solo los tokens, es lo que hace posible una alerta de pico de gasto más adelante.
Una dificultad común: olvidar los campos de tokens relacionados con la caché (cache_read_input_tokens) subestima o sobreestima el costo una vez que se utiliza el almacenamiento en caché de prompts.

Relacionado: Integración de paneles de uso y costos de Claude con Datadog - envío de este costo por llamada a un panel.

9. Registrar respuestas de streaming sin perder recuentos de tokens

El streaming complica el registro porque la respuesta llega en fragmentos.

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("Escribe una entrada de changelog de dos oraciones para una versión de corrección de errores.")

stream.get_final_message() te da el mensaje completo, incluidos los recuentos de usage finales, después de que el stream termina.
El registro solo ocurre una vez que el stream se consume por completo, por lo que la latencia de una llamada en streaming debe medirse hasta el último fragmento, no el primero.
Una dificultad común: registrar dentro del bucle for text in stream.text_stream se activaría una vez por fragmento en lugar de una vez por llamada, inundando tus registros.

Versiones de la pila: Escrito contra la línea de modelos de Claude actual a partir de ~junio de 2026: Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (el predeterminado) y Claude Haiku 4.5, y el SDK oficial de Python anthropic (última versión 0.x). Los nombres de los modelos, los precios y las versiones del SDK cambian rápidamente; verifica los detalles actuales en platform.claude.com/docs antes de confiar en ellos.

Conceptos básicos de observabilidad

9 ejemplos para empezar con la observabilidad: 6 básicos y 3 intermedios.

Prerrequisitos

Instala el SDK oficial: pip install anthropic.
Configura ANTHROPIC_API_KEY en tu entorno.
Los ejemplos solo usan la biblioteca estándar (logging, json, time, uuid) junto con anthropic, por lo que no se requieren paquetes de observabilidad adicionales para seguir el ejemplo.

Ejemplos básicos

1. Registrar el prompt y la respuesta de una sola llamada

La observabilidad mínima viable: registra lo que enviaste y lo que recibiste.

import anthropic
 
client = anthropic.Anthropic()
 
prompt = "Resume los riesgos de omitir la revisión 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"RESPUESTA: {response.content[0].text}")

response.content es una lista de bloques de contenido; [0].text obtiene el primer bloque de texto.
Este es el mínimo, no el máximo: una instrucción print está bien para un script único, pero no para producción.
Los siguientes ejemplos convierten esto en un registro estructurado y consultable en lugar de una salida de consola.

2. Capturar recuentos de tokens de la respuesta

Cada respuesta contiene datos de uso; deja de descartarlos.

import anthropic
 
client = anthropic.Anthropic()
 
response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=300,
    messages=[{"role": "user", "content": "Explica la idempotencia en un párrafo."}],
)
 
usage = response.usage
print(f"input_tokens={usage.input_tokens} output_tokens={usage.output_tokens}")

response.usage expone input_tokens y output_tokens en cada respuesta no streaming.
Los recuentos de tokens son la materia prima para el seguimiento de costos, así que captúralos incluso antes de crear un panel.
Una dificultad común: los recuentos de tokens son por respuesta, no acumulativos, por lo que una conversación de varios turnos requiere que los sumes tú mismo.

Relacionado: Registro estructurado de prompts, respuestas y recuentos de tokens - el esquema completo al que alimenta este campo.

3. Escribir una línea de registro JSON estructurada

Reemplaza las impresiones ad hoc con un objeto JSON por llamada.

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("Enumera tres beneficios de las feature flags.")

Una línea JSON por llamada es analizada trivialmente por cualquier agregador de registros (Datadog, CloudWatch, ELK) sin un analizador personalizado.
time.monotonic() evita problemas de ajuste del reloj que time.time() puede introducir para la medición de latencia.
Esta es una versión de una sola función del patrón que el artículo de registro estructurado formaliza en un esquema reutilizable.

4. Adjuntar un ID de solicitud para la correlación

Un ID de solicitud te permite vincular una línea de registro a un span de rastreo o a un ticket de soporte.

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("Redacta un mensaje de commit de una línea para una corrección de errores.")
print(result["request_id"])

Genera el ID de solicitud antes de la llamada para que exista incluso si la llamada genera una excepción.
Anthropic también devuelve su propio _request_id en el objeto de respuesta; registrar tu propio ID junto con él te permite correlacionar entre tus propios sistemas, así como con el soporte de Anthropic si es necesario.
Una dificultad común: olvidar propagar el ID de solicitud a cada línea de registro posterior (llamadas a herramientas, reintentos) rompe la correlación para la que lo creaste.

5. Registrar errores con el mismo esquema que los éxitos

Una llamada fallida todavía merece una entrada de registro estructurada, no solo un 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("Explica brevemente el teorema CAP.")

anthropic.APIStatusError cubre las respuestas 4xx/5xx de la API; captúrala explícitamente en lugar de un except Exception genérico.
Registrar errores con la misma forma JSON que los éxitos significa que una consulta de panel cubre ambos, en lugar de buscar en registros de errores separados.
Una dificultad común: tragar la excepción sin registrar status_code pierde la distinción entre un límite de tasa (429) y un error del servidor (500), que requieren respuestas diferentes.

6. Envolver el cliente para que cada llamada se registre automáticamente

Centraliza el registro en lugar de repetirlo en cada sitio de llamada.

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": "¿Qué es un filtro bloom?"}],
)

Envolver el cliente SDK significa que cada sitio de llamada obtiene el registro de forma gratuita, y no puedes olvidar agregarlo a una nueva llamada.
Esta es la semilla del patrón reutilizable utilizado en el resto de esta sección una vez que agregues spans de rastreo alrededor de create.
Una dificultad común: envolver solo messages.create omite las llamadas de streaming (messages.stream), que necesitan su propia instrumentación.

Relacionado: Cómo funciona la observabilidad para aplicaciones LLM - el modelo mental detrás de este patrón de envoltorio.

Ejemplos intermedios

7. Registrar cada turno en un bucle de agente de varios pasos

Un bucle de agente realiza varias llamadas por solicitud de usuario; registra cada una con un identificador compartido.

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})
        # Un agente real ejecutaría la llamada a la herramienta aquí y agregaría su resultado.
        break
 
    return "El bucle terminó sin una respuesta final."
 
run_agent_loop("¿Cuánto es 47 por 12? Muestra tu trabajo.")

El run_id compartido es lo que te permite agrupar posteriormente todos los pasos de una ejecución de agente en una única vista similar a un rastreo, incluso antes de agregar spans reales de OTel.
response.stop_reason te dice si el modelo quiere llamar a una herramienta (tool_use) o ha terminado (end_turn), lo cual es contexto esencial para la línea de registro.
Este registro plano por paso es el precursor de la estructura de spans anidados construida en el artículo de rastreo de OpenTelemetry.

Relacionado: Instrumentación de bucles de agente con rastreo de OpenTelemetry - convertir estos pasos de registro planos en spans anidados.

8. Calcular y registrar un costo estimado por llamada

Los recuentos de tokens por sí solos no te dicen el gasto; conviértelos con precios por modelo.

import json
import anthropic
 
client = anthropic.Anthropic()
 
# Tarifas ilustrativas solamente: verifica los precios actuales en 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("Da una definición de una oración de consistencia eventual.")

Los precios varían según el modelo y cambian con el tiempo, así que mantén la tabla de tarifas en un solo lugar y trátala como configuración, no como una constante codificada en los sitios de llamada.
Registrar el costo por llamada, no solo los tokens, es lo que hace posible una alerta de pico de gasto más adelante.
Una dificultad común: olvidar los campos de tokens relacionados con la caché (cache_read_input_tokens) subestima o sobreestima el costo una vez que se utiliza el almacenamiento en caché de prompts.

Relacionado: Integración de paneles de uso y costos de Claude con Datadog - envío de este costo por llamada a un panel.

9. Registrar respuestas de streaming sin perder recuentos de tokens

El streaming complica el registro porque la respuesta llega en fragmentos.

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("Escribe una entrada de changelog de dos oraciones para una versión de corrección de errores.")

stream.get_final_message() te da el mensaje completo, incluidos los recuentos de usage finales, después de que el stream termina.
El registro solo ocurre una vez que el stream se consume por completo, por lo que la latencia de una llamada en streaming debe medirse hasta el último fragmento, no el primero.
Una dificultad común: registrar dentro del bucle for text in stream.text_stream se activaría una vez por fragmento en lugar de una vez por llamada, inundando tus registros.

Versiones de la pila: Escrito contra la línea de modelos de Claude actual a partir de ~junio de 2026: Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (el predeterminado) y Claude Haiku 4.5, y el SDK oficial de Python anthropic (última versión 0.x). Los nombres de los modelos, los precios y las versiones del SDK cambian rápidamente; verifica los detalles actuales en platform.claude.com/docs antes de confiar en ellos.

Conceptos básicos de observabilidad

Buscar en la documentación

Conceptos básicos de observabilidad