Preludio
Construir un primer agente de IA de la manera difícil es un rito de paso. Llamadas a la API en bruto, estado de conversación manual, bucles de ejecución de herramientas escritos a mano, lógica de reintentos dispersa por tres archivos. Funciona, del mismo modo que funciona una casa construida sin planos. Se mantiene en pie. Pero tiene goteras.
Cuando Anthropic publicó el Agent SDK, ese mismo agente se reescribió en una tarde. No porque el SDK sea mágico. Porque gestiona las partes que todo agente necesita, el bucle agéntico, la ejecución de herramientas, la gestión de conversaciones, y te deja centrarte en las partes que hacen único a tu agente.
Esta guía construye un agente completo desde cero. No una demo que consulta el tiempo. Un agente de gestión de proyectos que crea tareas, comprueba estados, asigna trabajo, envía notificaciones y maneja los casos límite que las aplicaciones reales encuentran. Al final, tendrás un agente funcional, una comprensión de la arquitectura del SDK y patrones que puedes aplicar a cualquier agente que construyas.
El problema
Todo agente comparte el mismo desafío central. Necesitas dar a un LLM la capacidad de tomar acciones en el mundo real, manteniendo el control sobre cuáles son esas acciones, cómo se ejecutan y qué sucede cuando fallan.
El enfoque en bruto significa escribir el bucle agéntico tú mismo. Envía un mensaje a la API. Comprueba si la respuesta contiene llamadas a herramientas. Si es así, analízalas, ejecútalas, envía los resultados de vuelta y comprueba de nuevo. Maneja errores en cada paso.
Gestiona el historial de conversación. Controla el uso de tokens. Implementa timeouts. Añade logging.
Ese bucle no es complejo en principio. Es tedioso en la práctica. Y el tedio crea errores.
Un manejador de errores olvidado. Un historial de conversación que crece sin límite. Un parser de llamadas a herramientas que falla en casos límite.
El Agent SDK elimina el tedio. Te da una implementación probada y mantenida del bucle agéntico, y te permite definir las partes interesantes, las herramientas, las instrucciones, los guardrails, como simples funciones Python y configuración.
Así es como.
El camino: construye tu agente de IA paso a paso
Lo que vamos a construir
El agente que vamos a construir se llama TaskBot. Gestiona un tablero de proyecto simple con estas capacidades.
Puede crear tareas con título, descripción, prioridad y responsable. Puede listar tareas filtradas por estado o responsable. Puede actualizar el estado de una tarea. Puede enviar notificaciones cuando las tareas cambian. Y puede proporcionar resúmenes del progreso del proyecto.
Este es un caso de uso realista. Toca bases de datos, APIs externas y lógica de negocio. Requiere conversaciones multi-turno donde el usuario hace preguntas de seguimiento. Y necesita guardrails para prevenir el mal uso.
El código completo está al final de esta guía. Lo construiremos pieza por pieza para que entiendas cada decisión.
Configurando el proyecto
Empieza creando un directorio de proyecto e instalando el SDK.
mkdir taskbot && cd taskbot
python -m venv venv
source venv/bin/activate
pip install openai-agents anthropic
El Agent SDK se distribuye como el paquete openai-agents, que soporta múltiples proveedores de modelos incluyendo Claude a través de la integración con Anthropic. También necesitarás una clave API de Anthropic.
export ANTHROPIC_API_KEY="your-key-here"
Crea la estructura del proyecto.
taskbot/
agent.py # Agent definition
tools.py # Tool functions
guardrails.py # Input/output validation
models.py # Data models
database.py # Storage layer
main.py # Entry point
Esta estructura separa responsabilidades. Las herramientas son funciones puras que interactúan con la base de datos. La definición del agente es configuración. Los guardrails son lógica de validación. El punto de entrada une todo.
Definiendo los modelos de datos
Antes de escribir herramientas, define con qué vas a trabajar. Este ejemplo usa dataclasses de Python por simplicidad, pero los modelos de Pydantic también funcionan bien aquí.
# models.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
class TaskStatus(Enum):
TODO = "todo"
IN_PROGRESS = "in_progress"
REVIEW = "review"
DONE = "done"
class Priority(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class Task:
id: str
title: str
description: str
status: TaskStatus = TaskStatus.TODO
priority: Priority = Priority.MEDIUM
assignee: Optional[str] = None
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
Construyendo una capa de almacenamiento simple
Para este tutorial, una base de datos en memoria mantiene las cosas simples. En producción, la sustituirías por PostgreSQL, SQLite o lo que use tu aplicación. Al agente no le importa la implementación del almacenamiento. Solo interactúa con herramientas.
# database.py
from datetime import datetime
from models import Task, TaskStatus
from typing import Optional
import uuid
_tasks: dict[str, Task] = {}
def create_task(title: str, description: str,
priority: str = "medium",
assignee: Optional[str] = None) -> Task:
task_id = str(uuid.uuid4())[:8]
task = Task(
id=task_id,
title=title,
description=description,
priority=priority,
assignee=assignee
)
_tasks[task_id] = task
return task
def get_task(task_id: str) -> Optional[Task]:
return _tasks.get(task_id)
def list_tasks(status: Optional[str] = None,
assignee: Optional[str] = None) -> list[Task]:
tasks = list(_tasks.values())
if status:
tasks = [t for t in tasks if t.status.value == status]
if assignee:
tasks = [t for t in tasks if t.assignee == assignee]
return tasks
def update_task_status(task_id: str, new_status: str) -> Optional[Task]:
task = _tasks.get(task_id)
if task:
task.status = TaskStatus(new_status)
task.updated_at = datetime.now()
return task
Definiendo herramientas
Las herramientas son donde el agente se encuentra con el mundo real. Cada herramienta es una función Python decorada con @function_tool. El nombre de la función se convierte en el nombre de la herramienta. El docstring se convierte en la descripción de la herramienta, que Claude lee para entender cuándo y cómo usarla. Las anotaciones de tipo se convierten en el esquema de parámetros.
# tools.py
from agents import function_tool
from database import create_task, get_task, list_tasks, update_task_status
from typing import Optional
@function_tool
def create_new_task(title: str, description: str,
priority: str = "medium",
assignee: Optional[str] = None) -> str:
"""Create a new task on the project board.
Args:
title: Short title for the task
description: Detailed description of what needs to be done
priority: One of 'low', 'medium', 'high', or 'critical'
assignee: Name of the person to assign the task to
"""
task = create_task(title, description, priority, assignee)
return (
f"Created task {task.id}: '{task.title}' "
f"(priority: {priority}, assignee: {assignee or 'unassigned'})"
)
@function_tool
def list_project_tasks(status: Optional[str] = None,
assignee: Optional[str] = None) -> str:
"""List tasks on the project board, optionally filtered.
Args:
status: Filter by status ('todo', 'in_progress', 'review', 'done')
assignee: Filter by assignee name
"""
tasks = list_tasks(status, assignee)
if not tasks:
return "No tasks found matching the criteria."
lines = []
for t in tasks:
lines.append(
f"[{t.id}] {t.title} | "
f"Status: {t.status.value} | "
f"Priority: {t.priority} | "
f"Assignee: {t.assignee or 'unassigned'}"
)
return "\n".join(lines)
@function_tool
def update_status(task_id: str, new_status: str) -> str:
"""Update the status of an existing task.
Args:
task_id: The ID of the task to update
new_status: New status ('todo', 'in_progress', 'review', 'done')
"""
task = update_task_status(task_id, new_status)
if task:
return f"Updated task {task_id} to status '{new_status}'."
return f"Task {task_id} not found."
@function_tool
def get_task_details(task_id: str) -> str:
"""Get full details of a specific task.
Args:
task_id: The ID of the task to look up
"""
task = get_task(task_id)
if not task:
return f"Task {task_id} not found."
return (
f"Task {task.id}\n"
f"Title: {task.title}\n"
f"Description: {task.description}\n"
f"Status: {task.status.value}\n"
f"Priority: {task.priority}\n"
f"Assignee: {task.assignee or 'unassigned'}\n"
f"Created: {task.created_at.isoformat()}\n"
f"Updated: {task.updated_at.isoformat()}"
)
@function_tool
def send_notification(recipient: str, message: str) -> str:
"""Send a notification to a team member.
Args:
recipient: Name of the person to notify
message: The notification message
"""
# In production, this would call an email/Slack/Teams API
print(f"[NOTIFICATION to {recipient}]: {message}")
return f"Notification sent to {recipient}."
Observa que cada herramienta devuelve una cadena de texto. El SDK envía esta cadena de vuelta a Claude como resultado de la herramienta. Claude la usa para formular su respuesta. Valores de retorno claros e informativos mejoran las respuestas de Claude.
Los docstrings importan enormemente. Claude los lee para decidir qué herramienta usar y cómo llamarla. Un docstring vago produce un uso vago de la herramienta. Un docstring preciso con parámetros documentados produce llamadas precisas a herramientas.
Creando el agente
Con las herramientas definidas, el agente en sí es sencillo.
# agent.py
from agents import Agent
from tools import (
create_new_task,
list_project_tasks,
update_status,
get_task_details,
send_notification
)
taskbot = Agent(
name="TaskBot",
model="claude-sonnet-4-6",
instructions="""You are TaskBot, a project management assistant.
You help teams manage their tasks and stay organised.
Guidelines:
- When creating tasks, always confirm the details with the user first
- Use 'medium' priority unless the user specifies otherwise
- When updating task status, notify the assignee
- Provide concise summaries when listing tasks
- If a task ID is not found, suggest listing all tasks
- Always be helpful but never create tasks without explicit user request""",
tools=[
create_new_task,
list_project_tasks,
update_status,
get_task_details,
send_notification
]
)
El campo instructions es tu prompt de sistema. Da forma al comportamiento del agente en todas las conversaciones. Escríbelo como si estuvieras instruyendo a un nuevo miembro del equipo. Sé específico sobre lo que el agente debe y no debe hacer.
El campo model determina qué modelo de Claude impulsa al agente. Usa claude-sonnet-4-6 para un buen equilibrio entre velocidad y capacidad. Usa claude-opus-4-6 para tareas que requieren razonamiento más profundo. Usa claude-haiku-4-6 para tareas simples y de alto volumen donde la velocidad es lo más importante.
Entendiendo el bucle agéntico
Cuando llamas a Runner.run_sync(taskbot, "Create a task for the API migration"), esto es lo que sucede internamente.
Primero, el SDK envía tu mensaje a Claude junto con las instrucciones del sistema y las definiciones de herramientas. Claude recibe todo lo que necesita para entender quién es, qué puede hacer y qué quiere el usuario.
Segundo, Claude responde. Si decide llamar a una herramienta, la respuesta contiene un bloque de uso de herramienta con el nombre y los parámetros. Si decide responder directamente, la respuesta contiene texto.
Tercero, si hubo una llamada a herramienta, el SDK ejecuta tu función de herramienta con los parámetros proporcionados. Captura el valor de retorno como cadena.
Cuarto, el SDK envía el resultado de la herramienta de vuelta a Claude. Claude ahora sabe qué sucedió y puede decidir llamar a otra herramienta, hacer una pregunta de seguimiento o proporcionar una respuesta final.
Este bucle continúa hasta que Claude produce una respuesta sin llamadas a herramientas. Esa respuesta se convierte en el final_output de la ejecución.
La idea clave es que nunca escribes este bucle. El SDK lo gestiona. Tu trabajo es definir las herramientas e instrucciones que dan forma a lo que sucede dentro del bucle.
Conversaciones multi-turno
Una sola llamada a Runner.run_sync() gestiona un turno de conversación. Para interacciones multi-turno donde el usuario hace preguntas de seguimiento, necesitas mantener el historial de conversación.
# main.py
from agents import Runner
from agent import taskbot
async def chat():
history = []
while True:
user_input = input("\nYou: ")
if user_input.lower() in ("quit", "exit"):
break
# Add user message to history
history.append({
"role": "user",
"content": user_input
})
result = await Runner.run(
taskbot,
history
)
# Add assistant response to history
history.append({
"role": "assistant",
"content": result.final_output
})
print(f"\nTaskBot: {result.final_output}")
if __name__ == "__main__":
import asyncio
asyncio.run(chat())
El historial de conversación es una lista de objetos de mensaje. Cada mensaje tiene un role (user o assistant) y content. El SDK envía el historial completo con cada petición, dando a Claude el contexto de toda la conversación.
Ten en cuenta la longitud del historial. Cada mensaje consume tokens. Para sesiones de larga duración, puede que necesites resumir mensajes antiguos o implementar una ventana deslizante que mantenga solo los intercambios más recientes.
Añadiendo guardrails
Los guardrails son funciones que validan las entradas antes de que el agente las procese y las salidas antes de que el agente las devuelva. Son tu capa de seguridad.
# guardrails.py
from agents import (
InputGuardrail,
OutputGuardrail,
GuardrailFunctionOutput,
Agent,
Runner
)
BLOCKED_TERMS = [
"delete all", "drop table", "remove everything",
"fire everyone", "terminate all"
]
async def validate_input(ctx, agent, user_input):
"""Block potentially destructive or harmful requests."""
lower_input = user_input.lower() if isinstance(user_input, str) else ""
triggered = any(term in lower_input for term in BLOCKED_TERMS)
return GuardrailFunctionOutput(
output_info={
"blocked_term_found": triggered,
"input_length": len(lower_input)
},
tripwire_triggered=triggered
)
async def validate_output(ctx, agent, output):
"""Ensure the agent does not leak internal details."""
lower_output = output.lower() if isinstance(output, str) else ""
leaks_internals = any(
term in lower_output
for term in ["api_key", "database_url", "internal_secret"]
)
return GuardrailFunctionOutput(
output_info={"leaks_internals": leaks_internals},
tripwire_triggered=leaks_internals
)
Ahora añade los guardrails a la definición del agente.
from guardrails import validate_input, validate_output
from agents import InputGuardrail, OutputGuardrail
taskbot = Agent(
name="TaskBot",
model="claude-sonnet-4-6",
instructions="...",
tools=[...],
input_guardrails=[
InputGuardrail(guardrail_function=validate_input)
],
output_guardrails=[
OutputGuardrail(guardrail_function=validate_output)
]
)
Cuando un guardrail se activa, el SDK lanza una excepción que capturas en el código de tu aplicación. El agente nunca ve la entrada bloqueada. El usuario recibe un mensaje de error claro.
from agents.exceptions import InputGuardrailTripwireTriggered
try:
result = await Runner.run(taskbot, user_input)
except InputGuardrailTripwireTriggered:
print("That request was blocked by our safety policy.")
Recomendamos ejecutar guardrails en cada agente en producción. La sobrecarga es mínima, típicamente unos pocos milisegundos para coincidencia de patrones. La protección es significativa. Un guardrail que captura una petición destructiva se ha amortizado permanentemente.
Manejo de errores
Las herramientas fallan. Las APIs agotan su tiempo. Las bases de datos se caen. Tu agente necesita manejar estos fallos con elegancia.
El enfoque más simple es manejar los errores dentro de la propia función de herramienta.
@function_tool
def create_new_task(title: str, description: str,
priority: str = "medium",
assignee: Optional[str] = None) -> str:
"""Create a new task on the project board."""
try:
# Validate priority
valid_priorities = ["low", "medium", "high", "critical"]
if priority not in valid_priorities:
return (
f"Invalid priority '{priority}'. "
f"Must be one of: {', '.join(valid_priorities)}"
)
task = create_task(title, description, priority, assignee)
return (
f"Created task {task.id}: '{task.title}' "
f"(priority: {priority})"
)
except Exception as e:
return f"Failed to create task: {str(e)}"
Al devolver mensajes de error como cadenas en lugar de lanzar excepciones, dejas que Claude maneje el fallo conversacionalmente. Claude ve "Failed to create task: database connection timeout" y puede decirle al usuario qué sucedió, sugerir un reintento o probar un enfoque alternativo.
Para fallos críticos que deben detener completamente al agente, lanza una excepción. El SDK detendrá el bucle agéntico y propagará el error a tu código de aplicación.
También deberías establecer timeouts a nivel del runner para evitar que los agentes se ejecuten indefinidamente.
result = await Runner.run(
taskbot,
user_input,
max_turns=10 # Stop after 10 tool call cycles
)
El parámetro max_turns previene bucles infinitos donde el agente sigue llamando herramientas sin llegar a una conclusión. Diez turnos es un valor predeterminado razonable para la mayoría de los agentes. Auméntalo para agentes que necesitan realizar muchas operaciones secuenciales.
Handoffs entre agentes
A veces un solo agente no puede manejarlo todo. El Agent SDK soporta handoffs, donde un agente delega a otro para tareas especializadas.
Imagina que TaskBot necesita manejar tanto la gestión de proyectos como el seguimiento del tiempo. En lugar de meter ambas cosas en un solo agente, creas dos agentes especializados y dejas que se deleguen entre sí.
from agents import Agent
time_tracker = Agent(
name="TimeTracker",
model="claude-sonnet-4-6",
instructions="""You track time spent on tasks.
You can log hours, view time reports, and calculate
utilisation rates. For task management questions,
hand off to the TaskBot agent.""",
tools=[log_time, get_time_report, calculate_utilisation],
handoffs=[] # Will be set after taskbot is defined
)
taskbot = Agent(
name="TaskBot",
model="claude-sonnet-4-6",
instructions="""You manage project tasks.
For time tracking questions, hand off to the
TimeTracker agent.""",
tools=[
create_new_task,
list_project_tasks,
update_status,
get_task_details,
send_notification
],
handoffs=[time_tracker]
)
# Complete the circular reference
time_tracker.handoffs = [taskbot]
Cuando un usuario le pregunta a TaskBot "How many hours did Sarah log this week?", TaskBot reconoce que es una pregunta de seguimiento del tiempo y delega a TimeTracker. TimeTracker maneja la petición con sus herramientas especializadas y devuelve el resultado.
Este patrón mantiene a cada agente enfocado. Los agentes enfocados son más fáciles de probar, más fáciles de depurar y producen mejores resultados porque sus instrucciones y herramientas no se diluyen con capacidades no relacionadas.
Observabilidad y monitorización
En producción, necesitas saber qué está haciendo tu agente. El SDK proporciona hooks que te permiten observar cada paso del bucle agéntico.
from agents import RunHooks, RunContextWrapper, Tool, Agent
from datetime import datetime
class ProductionHooks(RunHooks):
def __init__(self):
self.tool_calls = []
self.start_time = None
self.total_tokens = 0
async def on_agent_start(self, context, agent):
self.start_time = datetime.now()
print(f"[{self.start_time}] Agent '{agent.name}' started")
async def on_tool_start(self, context, agent, tool):
print(f" [TOOL CALL] {tool.name}")
async def on_tool_end(self, context, agent, tool, result):
self.tool_calls.append({
"tool": tool.name,
"timestamp": datetime.now().isoformat(),
"result_length": len(str(result))
})
print(f" [TOOL DONE] {tool.name} ({len(str(result))} chars)")
async def on_agent_end(self, context, agent, output):
duration = (datetime.now() - self.start_time).total_seconds()
print(f"[COMPLETE] {len(self.tool_calls)} tool calls "
f"in {duration:.1f}s")
hooks = ProductionHooks()
result = await Runner.run(
taskbot,
user_input,
run_hooks=hooks
)
Estos hooks te proporcionan datos estructurados sobre cada ejecución del agente. En producción, envía estos datos a un servicio de logging para su análisis. Elementos comunes a rastrear incluyen el número de llamadas a herramientas por conversación, qué herramientas se usan con más frecuencia, el tiempo medio de ejecución, las tasas de error y los patrones de consumo de tokens.
Si estás construyendo agentes que se conectan a servicios externos a través de servidores MCP, la observabilidad se vuelve aún más importante. Las llamadas a herramientas MCP cruzan fronteras de red, por lo que necesitas rastrear la latencia y los fallos en cada salto.
Patrones de producción
Varios patrones han demostrado ser esenciales en agentes en producción.
Ejecución asíncrona. El SDK soporta async de forma nativa. Usa Runner.run() en lugar de Runner.run_sync() en producción para evitar bloquear el bucle de eventos de tu aplicación.
result = await Runner.run(taskbot, user_input)
Limitación de tasa. Si tu agente atiende a múltiples usuarios, implementa limitación de tasa para evitar el agotamiento de la cuota de la API.
import asyncio
from collections import defaultdict
from time import time
class RateLimiter:
def __init__(self, max_requests_per_minute=20):
self.max_rpm = max_requests_per_minute
self.requests = defaultdict(list)
async def check(self, user_id: str):
now = time()
# Clean old entries
self.requests[user_id] = [
t for t in self.requests[user_id]
if now - t < 60
]
if len(self.requests[user_id]) >= self.max_rpm:
raise Exception("Rate limit exceeded. Please wait.")
self.requests[user_id].append(now)
limiter = RateLimiter()
async def handle_request(user_id: str, message: str):
await limiter.check(user_id)
result = await Runner.run(taskbot, message)
return result.final_output
Seguimiento de costes. Cada llamada al agente consume tokens. Rastrea el uso para evitar sorpresas en la factura.
class CostTracker(RunHooks):
def __init__(self):
self.total_input_tokens = 0
self.total_output_tokens = 0
async def on_agent_end(self, context, agent, output):
usage = getattr(context, 'usage', None)
if usage:
self.total_input_tokens += usage.input_tokens
self.total_output_tokens += usage.output_tokens
cost = (
self.total_input_tokens * 0.003 / 1000 +
self.total_output_tokens * 0.015 / 1000
)
print(f"Session cost so far: ${cost:.4f}")
Probando tu agente
Probar agentes requiere dos capas. Tests unitarios para herramientas individuales y tests de integración para el bucle completo del agente.
Tests unitarios de herramientas. Como las herramientas son funciones Python normales (envueltas con un decorador), puedes probarlas directamente.
# test_tools.py
import pytest
from database import create_task, list_tasks, _tasks
def setup_function():
"""Clear the database before each test."""
_tasks.clear()
def test_create_task():
task = create_task(
"Fix login bug",
"Users cannot log in with SSO",
"high",
"Sarah"
)
assert task.title == "Fix login bug"
assert task.priority == "high"
assert task.assignee == "Sarah"
assert task.id is not None
def test_list_tasks_filter_by_status():
create_task("Task 1", "Description", "medium")
task2 = create_task("Task 2", "Description", "high")
task2.status = TaskStatus.DONE
todo_tasks = list_tasks(status="todo")
assert len(todo_tasks) == 1
assert todo_tasks[0].title == "Task 1"
def test_list_tasks_empty():
tasks = list_tasks()
assert tasks == []
Tests de integración del agente. Usa el SDK para ejecutar el agente contra entradas de prueba y verificar las salidas.
# test_agent.py
import pytest
from agents import Runner
from agent import taskbot
from database import _tasks
@pytest.fixture(autouse=True)
def clear_db():
_tasks.clear()
yield
_tasks.clear()
@pytest.mark.asyncio
async def test_agent_creates_task():
result = await Runner.run(
taskbot,
"Create a high priority task called 'Deploy v2.0' "
"about deploying the new version to production"
)
assert len(_tasks) == 1
task = list(_tasks.values())[0]
assert "Deploy" in task.title
@pytest.mark.asyncio
async def test_agent_handles_unknown_task():
result = await Runner.run(
taskbot,
"Show me details for task xyz123"
)
assert "not found" in result.final_output.lower()
@pytest.mark.asyncio
async def test_guardrail_blocks_destructive_input():
from agents.exceptions import InputGuardrailTripwireTriggered
with pytest.raises(InputGuardrailTripwireTriggered):
await Runner.run(taskbot, "Delete all tasks immediately")
Los tests de integración son más lentos porque hacen llamadas a la API. Ejecútalos en una suite de tests separada y usa variables de entorno para apuntar a una clave API de prueba con límites de tasa más bajos.
El ejemplo completo funcional
Aquí está el agente TaskBot completo en un solo archivo, listo para ejecutar.
#!/usr/bin/env python3
"""TaskBot - A project management agent built with the Claude Agent SDK."""
import asyncio
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
from agents import (
Agent,
Runner,
function_tool,
InputGuardrail,
GuardrailFunctionOutput,
RunHooks,
)
# --- Data Models ---
@dataclass
class Task:
id: str
title: str
description: str
status: str = "todo"
priority: str = "medium"
assignee: Optional[str] = None
created_at: str = field(
default_factory=lambda: datetime.now().isoformat()
)
# --- Database ---
tasks_db: dict[str, Task] = {}
# --- Tools ---
@function_tool
def create_task(title: str, description: str,
priority: str = "medium",
assignee: Optional[str] = None) -> str:
"""Create a new task on the project board.
Args:
title: Short title for the task
description: What needs to be done
priority: low, medium, high, or critical
assignee: Person to assign the task to
"""
valid = ["low", "medium", "high", "critical"]
if priority not in valid:
return f"Invalid priority. Must be one of: {', '.join(valid)}"
task_id = str(uuid.uuid4())[:8]
task = Task(
id=task_id, title=title, description=description,
priority=priority, assignee=assignee
)
tasks_db[task_id] = task
return (
f"Created task {task_id}: '{title}' "
f"(priority: {priority}, "
f"assignee: {assignee or 'unassigned'})"
)
@function_tool
def list_tasks(status: Optional[str] = None,
assignee: Optional[str] = None) -> str:
"""List tasks, optionally filtered by status or assignee.
Args:
status: Filter by todo, in_progress, review, or done
assignee: Filter by assignee name
"""
filtered = list(tasks_db.values())
if status:
filtered = [t for t in filtered if t.status == status]
if assignee:
filtered = [t for t in filtered if t.assignee == assignee]
if not filtered:
return "No tasks found."
lines = []
for t in filtered:
lines.append(
f"[{t.id}] {t.title} | {t.status} | "
f"{t.priority} | {t.assignee or 'unassigned'}"
)
return "\n".join(lines)
@function_tool
def update_task(task_id: str, new_status: str) -> str:
"""Update the status of a task.
Args:
task_id: The task ID
new_status: New status (todo, in_progress, review, done)
"""
valid = ["todo", "in_progress", "review", "done"]
if new_status not in valid:
return f"Invalid status. Must be one of: {', '.join(valid)}"
task = tasks_db.get(task_id)
if not task:
return f"Task {task_id} not found."
old_status = task.status
task.status = new_status
return (
f"Updated task {task_id} from '{old_status}' "
f"to '{new_status}'."
)
@function_tool
def get_task(task_id: str) -> str:
"""Get full details of a task.
Args:
task_id: The task ID to look up
"""
task = tasks_db.get(task_id)
if not task:
return f"Task {task_id} not found."
return (
f"Task: {task.id}\n"
f"Title: {task.title}\n"
f"Description: {task.description}\n"
f"Status: {task.status}\n"
f"Priority: {task.priority}\n"
f"Assignee: {task.assignee or 'unassigned'}\n"
f"Created: {task.created_at}"
)
@function_tool
def notify(recipient: str, message: str) -> str:
"""Send a notification to a team member.
Args:
recipient: Person to notify
message: Notification message
"""
print(f" [NOTIFY {recipient}]: {message}")
return f"Notification sent to {recipient}."
# --- Guardrails ---
async def check_input(ctx, agent, user_input):
blocked = ["delete all", "remove everything", "drop table"]
text = user_input.lower() if isinstance(user_input, str) else ""
triggered = any(term in text for term in blocked)
return GuardrailFunctionOutput(
output_info={"blocked": triggered},
tripwire_triggered=triggered
)
# --- Hooks ---
class AgentLogger(RunHooks):
async def on_tool_start(self, context, agent, tool):
print(f" > Calling {tool.name}...")
async def on_tool_end(self, context, agent, tool, result):
preview = str(result)[:80]
print(f" < {tool.name} returned: {preview}")
# --- Agent ---
taskbot = Agent(
name="TaskBot",
model="claude-sonnet-4-6",
instructions="""You are TaskBot, a project management assistant.
Rules:
- Confirm details before creating tasks
- Default to medium priority unless told otherwise
- Notify assignees when their tasks change status
- Be concise and helpful
- Never create tasks without an explicit request""",
tools=[create_task, list_tasks, update_task, get_task, notify],
input_guardrails=[
InputGuardrail(guardrail_function=check_input)
]
)
# --- Main ---
async def main():
print("TaskBot ready. Type 'quit' to exit.\n")
history = []
hooks = AgentLogger()
while True:
user_input = input("You: ")
if user_input.lower() in ("quit", "exit"):
break
history.append({"role": "user", "content": user_input})
try:
result = await Runner.run(
taskbot, history,
run_hooks=hooks, max_turns=10
)
response = result.final_output
history.append(
{"role": "assistant", "content": response}
)
print(f"\nTaskBot: {response}\n")
except Exception as e:
print(f"\nError: {e}\n")
if __name__ == "__main__":
asyncio.run(main())
Guarda esto como taskbot.py, configura tu ANTHROPIC_API_KEY y ejecútalo con python taskbot.py. Tendrás un agente de gestión de proyectos funcional que crea tareas, rastrea estados, envía notificaciones y bloquea entradas destructivas.
La lección
Construir un agente no se trata de la biblioteca. Se trata de las decisiones que tomas al usarla. Qué herramientas exponer. Qué instrucciones escribir.
Dónde poner guardrails. Cómo manejar los fallos.
El Agent SDK gestiona las partes mecánicas, el bucle agéntico, la ejecución de herramientas, la gestión de conversaciones, para que puedas centrarte en estas decisiones. Cada hora que antes se gastaba depurando un bucle escrito a mano es ahora una hora dedicada a mejorar herramientas e instrucciones.
Los patrones de esta guía se transfieren a cualquier agente que construyas. Las herramientas serán diferentes. El dominio será diferente. Pero la estructura, definiciones claras de herramientas con docstrings precisos, instrucciones reflexivas, guardrails en los límites, observabilidad en todo el sistema, permanece igual.
Para una comparación de cómo el Agent SDK se compara con otros frameworks, nuestra guía sobre Claude Agent SDK vs LangChain cubre esto en detalle. Y para extender tus agentes con servicios externos, la guía sobre servidores MCP y extensiones muestra cómo conectar agentes al ecosistema más amplio de herramientas.
Conclusión
Esta guía empezó describiendo un primer agente construido de la manera difícil con llamadas a la API en bruto y bucles manuales. TaskBot es el mismo tipo de agente, pero construido en una tarde en lugar de una semana. Tiene guardrails, observabilidad, manejo de errores y soporte multi-turno.
Es testable. Es mantenible. Y el código es lo suficientemente legible para que un nuevo miembro del equipo pueda entenderlo sin una explicación guiada.
El Agent SDK no hace nada que no pudieras hacer tú mismo. Hace lo que tú harías, pero probado, mantenido y mejorado por el equipo que construye Claude. Esa es la propuesta de valor. No magia. Apalancamiento.
Empieza con una sola herramienta. Hazla funcionar. Añade una segunda. Añade guardrails. Añade observabilidad.
Cada paso es pequeño. Cada paso hace tu agente más capaz y más fiable. Y cada agente que construyes después del primero es más rápido, porque los patrones son los mismos.
Construye algo. Ponlo en producción. Observa cómo tus usuarios interactúan con él. Luego mejóralo. Ese es el bucle que importa más que cualquier bucle agéntico en cualquier SDK.
