Cómo construir un agente productivo con el SDK de Anthropic, paso a paso
Construir un agente sólido con el SDK de Anthropic ya no requiere pegamento artesanal. Este tutorial monta paso a paso un agente productivo en Python: tool use, streaming con backpressure, prompt caching, un servidor MCP propio y trazas OTel. El resultado es un repositorio que puedes desplegar en un contenedor pequeño y operar sin sorpresas. Ver también: la guía completa del mcp model context protocol para el contexto de protocolo en el que encaja la pieza MCP de este tutorial.
Puntos clave
- El SDK de Anthropic separa dos capas: la Messages API del paquete
anthropicpara tool use, streaming y caching, y el Claude Agent SDK (claude-agent-sdk) para registrar servidores MCP y aplicar política conallowed_tools. - El bucle de tool use es siempre el mismo:
messages.createcontools=[...], leerstop_reason, ejecutar eltool_use, devolver untool_resulty volver a llamar hastaend_turn. - El streaming exige pensar en backpressure desde el primer prototipo: una
asyncio.Queueacotada entre eltext_streamy el consumidor evita que el SDK se bloquee y que la UI quede atrás. - El prompt caching paga cuando hay un bloque grande de contexto reusado; con
cache_control={"type": "ephemeral"}, los cache hits cuestan el 10% del input base, según la documentación oficial de Anthropic. - La observabilidad ya está estandarizada: las convenciones GenAI de OpenTelemetry definen el atributo
gen_ai.provider.name="anthropic"y operacioneschatyexecute_toolque cualquier backend OTLP entiende. - El empaquetado en un Dockerfile multi-stage con
python:3.12-slimy usuario no-root deja el agente listo para correr en cualquier runtime de contenedores moderno.
Requisitos previos y estructura del proyecto
Necesitas Python 3.12 o superior, una cuenta en Anthropic con ANTHROPIC_API_KEY, Docker para el último paso y, opcionalmente, un collector de OpenTelemetry local (Tempo, Jaeger u otro) para inspeccionar las trazas durante el desarrollo. Si vienes del modelo donde el agente conduce directamente la interfaz del sistema operativo, conviene contrastar este enfoque con Computer Use de Claude: aquí trabajamos con tool use estructurado y MCP, mucho más auditable.
La estructura del repositorio que vamos a montar separa con claridad las capas: la lógica del agente vive en app/, el servidor MCP propio en mcp_server/, los tests en tests/ y la configuración operativa en ops/. La imagen de cabecera muestra el árbol completo y sirve de mapa para el resto del post.
agente-anthropic-sdk/
├── app/
│ ├── main.py
│ ├── streaming.py
│ ├── caching.py
│ ├── observability.py
│ └── tools/
├── mcp_server/
│ ├── server.py
│ └── tools.py
├── tests/
├── ops/
├── Dockerfile
└── requirements.txtLas dependencias mínimas son cinco. La librería anthropic cubre la Messages API; claude-agent-sdk añade la integración MCP y la política allowed_tools; mcp se usa para el servidor propio; opentelemetry-sdk y el exportador OTLP gRPC instrumentan el conjunto.
anthropic>=0.40,<0.50
claude-agent-sdk>=0.3,<1.0
mcp>=1.6,<2.0
opentelemetry-sdk>=1.27,<2.0
opentelemetry-exporter-otlp-proto-grpc>=1.27,<2.0
python-dotenv>=1.0,<2.0Pinea las versiones a un rango compatible y no más; el ecosistema se mueve rápido y un pin estricto envejece mal en pocas semanas. Carga las variables de entorno al arrancar con python-dotenv desde un fichero .env que nunca se versiona.
Primer turno con tool use
El bucle de tool use es el corazón del agente. Defines una herramienta con su input_schema JSON Schema, la registras en la llamada a messages.create y procesas los bloques tool_use que aparezcan en la respuesta. Cuando termines de ejecutar la función local, devuelves el resultado como tool_result y vuelves a llamar al modelo con la conversación acumulada. La documentación del SDK de Anthropic[1] lo formaliza así: el modelo responde con stop_reason: "tool_use", tu código ejecuta la operación y mandas un tool_result para cerrar el ciclo.
# app/main.py
import anthropic
client = anthropic.Anthropic()
TOOLS = [
{
"name": "get_weather",
"description": "Obtiene la temperatura actual de una ciudad.",
"input_schema": {
"type": "object",
"properties": {
"location": {"type": "string", "description": "Ciudad y país."}
},
"required": ["location"],
},
}
]
def run_tool(name: str, args: dict) -> str:
if name == "get_weather":
# Sustituye por tu integración real (Open-Meteo, AEMET, etc.).
return f"21C, despejado en {args['location']}"
raise ValueError(f"Herramienta desconocida: {name}")
def chat(prompt: str) -> str:
messages = [{"role": "user", "content": prompt}]
while True:
response = client.messages.create(
model="claude-opus-4-7",
max_tokens=1024,
tools=TOOLS,
messages=messages,
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
return "".join(b.text for b in response.content if b.type == "text")
tool_results = []
for block in response.content:
if block.type == "tool_use":
output = run_tool(block.name, block.input)
tool_results.append(
{"type": "tool_result", "tool_use_id": block.id, "content": output}
)
messages.append({"role": "user", "content": tool_results})
Hay tres detalles que conviene fijar desde el primer turno. El bucle termina cuando stop_reason es distinto de tool_use; cualquier otra cosa significa que el modelo aún quiere ejecutar herramientas. El historial de mensajes se acumula, no se reescribe: el modelo necesita ver tanto su propio tool_use como el tool_result correspondiente para razonar el siguiente paso. Y los errores de la herramienta no se silencian — si la integración real lanza una excepción, devuelve un tool_result con is_error: True y un mensaje legible para que el modelo pueda reaccionar y reintentar o pedir ayuda.
Streaming y backpressure
Para una UI conversacional o un canal websocket, usar streaming en lugar de bloquear hasta el final es la diferencia entre una experiencia fluida y una que parece colgada. El SDK expone client.messages.stream(...) como context manager, y el atributo text_stream itera sobre los chunks de texto a medida que llegan.
# app/streaming.py
import asyncio
import anthropic
client = anthropic.Anthropic()
async def stream_to_consumer(prompt: str, queue: asyncio.Queue) -> None:
# El consumidor debe pasar una cola acotada, p. ej. asyncio.Queue(maxsize=50);
# una cola sin maxsize anula el contraflujo.
try:
with client.messages.stream(
model="claude-opus-4-7",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
) as stream:
for text in stream.text_stream:
# El timeout protege contra consumidores que no leen.
await asyncio.wait_for(queue.put(text), timeout=2.0)
except asyncio.TimeoutError:
# Backpressure: descarta o devuelve 503 al cliente final.
await queue.put(None)
except anthropic.APIStatusError as exc:
await queue.put(f"[error {exc.status_code}]")
raise
El bloque with garantiza que el stream se cierra aunque el consumidor lance una excepción. La asyncio.Queue(maxsize=N) acotada es el truco que evita el problema más común con streaming: si el cliente final consume despacio, el productor llena memoria sin que el SDK se entere. Con wait_for y un timeout corto, el productor detecta antes la presión y degrada con elegancia, devolviendo un 503 al cliente final o cortando la generación con stream.close().
Tres errores merecen tratamiento explícito. anthropic.APIStatusError con código 529 indica sobrecarga del servicio y es candidato a reintento con backoff. La cancelación del cliente (websocket cerrado) debe propagarse cancelando la tarea que lee de text_stream, no esperando a que termine. Y los timeouts intermedios — un proxy que corta a los 60 segundos — son distintos de los 529 y se gestionan reabriendo el stream desde el último punto coherente, no reintentando el turno completo.
Prompt caching: cuándo paga la pena
El prompt caching convierte los tokens de input estables en lectura barata. Cuando marcas un bloque del prompt con cache_control={"type": "ephemeral"}, Anthropic guarda esa porción y, en peticiones posteriores que la reutilicen, te cobra por ella un 10% del precio base de input — un descuento del 90% sobre cache hit, según la documentación del SDK de Anthropic[1] en su página de prompt caching. La primera escritura tiene un sobrecoste (1.25× para TTL de 5 minutos, 2× para TTL de 1 hora), así que la regla del codo es clara: cachea bloques que vayas a reutilizar al menos tres veces.
# app/caching.py
import anthropic
client = anthropic.Anthropic()
LARGE_SYSTEM = """Eres un asistente de soporte de jacar.es. Sigue estas reglas...
[... varios miles de tokens de instrucciones, ejemplos few-shot y políticas ...]
"""
def ask(history: list[dict], user_msg: str) -> str:
response = client.messages.create(
model="claude-opus-4-7",
max_tokens=1024,
system=[
{
"type": "text",
"text": LARGE_SYSTEM,
"cache_control": {"type": "ephemeral"},
}
],
messages=history + [{"role": "user", "content": user_msg}],
)
usage = response.usage
print(
"input:", usage.input_tokens,
"cache_read:", getattr(usage, "cache_read_input_tokens", 0),
"cache_write:", getattr(usage, "cache_creation_input_tokens", 0),
)
return response.content[0].text
La pista de que estás cacheando bien aparece en response.usage: tras el primer turno, cache_creation_input_tokens cuenta la escritura; en los siguientes, cache_read_input_tokens se infla y input_tokens se queda en lo que cambió. Si nunca ves lecturas de caché, alguno de los tokens del bloque cambió entre llamadas — un timestamp dinámico, una lista en orden distinto — y el cache key se invalida. Cachea bloques estables, fija el orden de los ejemplos y deja la parte volátil fuera del bloque marcado.
Dos casos donde el caching no compensa: prompts cortos con poca reutilización (la escritura inicial pesa más que el ahorro acumulado) y agentes que cambian de system prompt en cada turno por personalización extrema. En esos escenarios, deja cache_control fuera y revisa la arquitectura: probablemente haya margen para mover personalizaciones a un bloque user fijo y dejar el system estable.
Registrar un servidor MCP propio
El Claude Agent SDK expone el registro de servidores MCP como API de primera clase, una pieza clave que conecta este tutorial con la guía completa del mcp model context protocol. En Python, la integración pasa por claude_agent_sdk.query() y ClaudeAgentOptions(mcp_servers=..., allowed_tools=...). La convención de nombres mcp__<server-name>__<tool-name> ya estandariza cómo el modelo ve cada herramienta sin colisiones.
Empezamos por el servidor. Una herramienta create_ticket que abre una incidencia en el dominio del producto, expuesta vía stdio para que el agente la lance como subproceso:
# mcp_server/server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("dominio")
@mcp.tool()
def create_ticket(title: str, priority: str = "normal") -> dict:
"""Crea un ticket en el sistema interno. Devuelve {id, url}."""
# Sustituye por tu integración real (Linear, Jira, sistema propio).
return {"id": "TCK-1042", "url": "https://soporte.example.com/tickets/1042"}
if __name__ == "__main__":
mcp.run() # stdio por defecto
Y el cliente que lo consume desde el agente, registrando el servidor por código y limitando con allowed_tools exactamente la herramienta que queremos exponer al modelo:
# app/main.py (variante con MCP)
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
async def run_agent(prompt: str) -> None:
options = ClaudeAgentOptions(
mcp_servers={
"dominio": {
"command": "python",
"args": ["-m", "mcp_server.server"],
}
},
allowed_tools=["mcp__dominio__create_ticket"],
)
async for message in query(prompt=prompt, options=options):
if isinstance(message, ResultMessage) and message.subtype == "success":
print(message.result)
asyncio.run(run_agent("Abre un ticket de prioridad alta para el incidente del checkout."))
Tres reglas aprendidas en producción. Las credenciales se inyectan con env, nunca en el prompt: env={"DOMAIN_API_TOKEN": os.environ["DOMAIN_API_TOKEN"]} en el descriptor pone la rotación bajo control del despliegue. La política se afina con allowed_tools por nombre exacto cuando la lista es manejable y con mcp__dominio__* cuando confías en todo el servidor; el wildcard global mcp__* es casi siempre demasiado. Y los tests de contrato sobre el listado de herramientas — un test_mcp_contract.py que enumera y compara contra un snapshot — atrapan renames y removals antes de que rompan el agente en producción.
Observabilidad con OTel GenAI
Las convenciones semánticas GenAI de OpenTelemetry[2] ya cubren el caso Anthropic con el atributo gen_ai.provider.name="anthropic", el nombre de span chat {model} y operaciones chat, execute_tool y embeddings. Eso significa que tus trazas cruzan en Grafana o Honeycomb con el resto de la traza distribuida sin esfuerzo. La sibling cluster sobre observabilidad de agentes con OTel GenAI profundiza el detalle; aquí basta con cablear el exportador y abrir un span por cada llamada al modelo.
# app/observability.py
from contextlib import contextmanager
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agente-anthropic-sdk")
@contextmanager
def chat_span(model: str):
with tracer.start_as_current_span(f"chat {model}") as span:
span.set_attribute("gen_ai.provider.name", "anthropic")
span.set_attribute("gen_ai.system", "Anthropic") # compat con instrumentaciones más antiguas
span.set_attribute("gen_ai.operation.name", "chat")
span.set_attribute("gen_ai.request.model", model)
try:
yield span
except Exception as exc:
span.record_exception(exc)
span.set_status(trace.Status(trace.StatusCode.ERROR))
raise
Emitimos también gen_ai.system para que los colectores y herramientas que arrancaron con la spec inicial sigan reconociendo el agente sin reconfigurarse.
Envuelve cada messages.create con with chat_span("claude-opus-4-7") as span: y, después de la llamada, añade los atributos de uso desde response.usage: gen_ai.usage.input_tokens, gen_ai.usage.output_tokens y, si llevas caching, gen_ai.usage.input_tokens.cache_read. El exportador OTLPSpanExporter() toma el endpoint de la variable OTEL_EXPORTER_OTLP_ENDPOINT, así que apuntando al collector local durante el desarrollo y a Tempo o Honeycomb en producción no toca código. Ver también: la especificación oficial de MCP[3] si quieres correlar las trazas del agente con las del servidor MCP usando los atributos mcp.method.name y mcp.session.id.
Empaquetado y despliegue
Un Dockerfile multi-stage con python:3.12-slim da una imagen pequeña, reproducible y razonablemente segura. La idea: instalar dependencias en una etapa intermedia con caché de pip y copiar solo el resultado más el código en la etapa final, que corre como usuario no-root.
# Dockerfile FROM python:3.12-slim AS builder WORKDIR /build COPY requirements.txt . RUN pip install --no-cache-dir --target /opt/deps -r requirements.txt FROM python:3.12-slim RUN useradd --create-home --uid 10001 agente WORKDIR /app COPY --from=builder /opt/deps /opt/deps COPY app/ ./app/ COPY mcp_server/ ./mcp_server/ ENV PYTHONPATH=/opt/deps PYTHONUNBUFFERED=1 USER agente ENTRYPOINT ["python", "-m", "app.main"]
Tres decisiones que importan más de lo que parece. El usuario agente con UID fijo 10001 evita correr como root y simplifica políticas de PodSecurity en Kubernetes. PYTHONPATH=/opt/deps permite mantener el código del proyecto fuera del directorio de site-packages, lo que ayuda a separar diagnósticos. Y PYTHONUNBUFFERED=1 hace que los logs salgan inmediatamente, fundamental para que un orquestador detecte fallos de arranque sin retrasos artificiales. Si vas a ejecutar también el servidor MCP en el contenedor, declara un ENTRYPOINT distinto para esa imagen o usa un sidecar; en clústers más grandes, los patrones de sistemas multi-agente con LangGraph, CrewAI y AutoGen describen cómo tratar este contenedor como un nodo de un grafo de agentes.
En CI, ejecuta los tests con una versión fija de Python — el GIL opcional de Python 3.13 cambia los supuestos de concurrencia y conviene fijar la versión hasta validar el comportamiento del SDK con nogil. Los tests de contrato del servidor MCP, los snapshots del listado de herramientas y un test mínimo del bucle de tool use con un mock del cliente cubren las regresiones más típicas.
Conclusión
Un agente productivo con el SDK de Anthropic se reduce a seis piezas bien aisladas: bucle de tool use, streaming con backpressure, prompt caching donde paga, servidor MCP propio, trazas OTel y un contenedor pequeño. Cada pieza es ortogonal a las demás, lo que facilita iterar sin romper el conjunto y revisarlas por separado en code review. La consecuencia operativa es la que vale: un agente que se entiende, se observa y se despliega como cualquier otro servicio. Tienes el repositorio de referencia con todo el código de este tutorial en github.com/jacarsystems/jacar-anthropic-sdk-demo[4] — clónalo, ajusta el .env y arranca con docker compose up.
Síguenos en jacar.es para más sobre agentes en producción, MCP, observabilidad GenAI y patrones reales del ecosistema Anthropic.
