Building AI Agents That Actually Work in Production
·6 min read·AI AgentsLLMClaude AIPythonTypeScriptAutomation
AI agents have moved from demo territory to production. Companies are shipping agents that write code, manage workflows, process documents, and coordinate with external APIs — autonomously. But the gap between an agent that works in a demo and one that runs reliably under real conditions is significant.
This post covers the architecture, patterns, and failure modes that matter when you're building agents for production.
What an Agent Actually Is
An agent is a loop where the model decides what to do next. The loop continues until the task is complete or a stop condition is reached.
User task
↓
Model reasons about task
↓
Model calls a tool (or finishes)
↓
Tool executes, result returned
↓
Model reasons about result
↓
... repeat until done
The model is the reasoning engine. Your job is to define the tools, manage the context, handle errors, and decide when to stop.
The Minimal Agent Loop
import anthropic
import json
from typing import Any
client = anthropic.Anthropic()
def run_agent(
task: str,
tools: list[dict],
tool_executor: dict[str, callable],
max_steps: int = 20
) -> str:
messages = [{"role": "user", "content": task}]
for step in range(max_steps):
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=4096,
tools=tools,
messages=messages
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason == "end_turn":
text_blocks = [b for b in response.content if b.type == "text"]
return text_blocks[-1].text if text_blocks else ""
if response.stop_reason == "tool_use":
results = []
for block in response.content:
if block.type != "tool_use":
continue
try:
fn = tool_executor.get(block.name)
if fn is None:
raise ValueError(f"Unknown tool: {block.name}")
result = fn(**block.input)
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": json.dumps(result)
})
except Exception as e:
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": f"Error: {str(e)}",
"is_error": True
})
messages.append({"role": "user", "content": results})
return f"Agent stopped after {max_steps} steps without completing task"Designing Tools
Tool design is where most agents succeed or fail. The model can only use what you define, and poorly designed tools lead to bad decisions.
Rule 1: One tool, one responsibility
# ❌ Too broad — model can't reason about what it does
{
"name": "database_operations",
"description": "Do things with the database",
"input_schema": {
"type": "object",
"properties": {
"operation": {"type": "string"},
"query": {"type": "string"}
}
}
}
# ✅ Specific — model knows exactly when to use it
{
"name": "search_users",
"description": "Search users by name or email. Returns up to 10 results.",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Name or email to search for"
},
"limit": {
"type": "integer",
"description": "Max results (1-10)",
"default": 5
}
},
"required": ["query"]
}
}Rule 2: Descriptions are prompts
The description field is what the model reads to decide whether to call the tool. Write it like documentation — include what the tool returns, edge cases, and when NOT to use it.
{
"name": "execute_sql",
"description": """Execute a SELECT query against the production database.
Returns rows as a list of dicts.
Use this for data retrieval only — does not support INSERT, UPDATE, or DELETE.
Limit results with LIMIT clause for large tables."""
}Rule 3: Return structured data
# ❌ Unstructured — model has to parse prose
def get_user(user_id: str) -> str:
user = db.get_user(user_id)
return f"User {user.name} was created on {user.created_at}"
# ✅ Structured — model can extract what it needs
def get_user(user_id: str) -> dict:
user = db.get_user(user_id)
return {
"id": user.id,
"name": user.name,
"email": user.email,
"created_at": user.created_at.isoformat(),
"status": user.status
}Multi-Agent Patterns
For complex tasks, one agent is often not enough. Multi-agent patterns let you parallelize work and specialize roles.
Orchestrator → Subagent
def orchestrator(task: str) -> str:
# Orchestrator plans and delegates
plan = client.messages.create(
model="claude-opus-4-7", # smarter model for planning
max_tokens=1024,
system="Break the task into subtasks and assign each to a specialist agent.",
messages=[{"role": "user", "content": task}]
)
subtasks = parse_plan(plan.content[0].text)
results = []
for subtask in subtasks:
agent_type = subtask["agent"]
result = run_specialist_agent(agent_type, subtask["task"])
results.append(result)
# Synthesize results
return synthesize(results)Parallel Agents
import asyncio
async def run_agents_parallel(tasks: list[str]) -> list[str]:
async def run_one(task):
return await run_agent_async(task, tools, tool_executor)
return await asyncio.gather(*[run_one(t) for t in tasks])
# Analyze 5 documents simultaneously
results = asyncio.run(run_agents_parallel([
"Analyze Q1 sales report",
"Analyze Q2 sales report",
"Analyze Q3 sales report",
"Analyze Q4 sales report",
"Analyze competitor pricing doc"
]))Handling Failures
Agents that work in demos break in production because production has errors. Build failure handling in from the start.
from dataclasses import dataclass
from enum import Enum
class AgentStatus(Enum):
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
MAX_STEPS_REACHED = "max_steps_reached"
@dataclass
class AgentResult:
status: AgentStatus
output: str | None
error: str | None
steps: int
tool_calls: list[dict]
def run_agent_safe(task: str, **kwargs) -> AgentResult:
tool_calls = []
steps = 0
try:
result = run_agent(task, **kwargs)
return AgentResult(
status=AgentStatus.COMPLETED,
output=result,
error=None,
steps=steps,
tool_calls=tool_calls
)
except anthropic.RateLimitError:
return AgentResult(
status=AgentStatus.FAILED,
output=None,
error="Rate limited — retry after backoff",
steps=steps,
tool_calls=tool_calls
)
except Exception as e:
return AgentResult(
status=AgentStatus.FAILED,
output=None,
error=str(e),
steps=steps,
tool_calls=tool_calls
)Implement circuit breakers for external tools:
from functools import wraps
import time
class CircuitBreaker:
def __init__(self, failures_threshold=3, timeout=60):
self.failures = 0
self.threshold = failures_threshold
self.timeout = timeout
self.opened_at = None
def __call__(self, fn):
@wraps(fn)
def wrapper(*args, **kwargs):
if self.opened_at:
if time.time() - self.opened_at < self.timeout:
raise Exception(f"Circuit open for {fn.__name__}")
self.opened_at = None
self.failures = 0
try:
result = fn(*args, **kwargs)
self.failures = 0
return result
except Exception as e:
self.failures += 1
if self.failures >= self.threshold:
self.opened_at = time.time()
raise
return wrapper
breaker = CircuitBreaker(failures_threshold=3, timeout=30)
@breaker
def call_external_api(payload):
# This will open the circuit after 3 consecutive failures
return requests.post("https://api.example.com/data", json=payload)Observability
You can't debug what you can't see. Log every step of the agent loop:
import logging
import uuid
from datetime import datetime
logger = logging.getLogger("agent")
def run_agent_with_tracing(task: str, **kwargs) -> AgentResult:
trace_id = str(uuid.uuid4())[:8]
start = datetime.utcnow()
logger.info(f"[{trace_id}] Starting agent task: {task[:100]}")
messages = [{"role": "user", "content": task}]
step = 0
for step in range(kwargs.get("max_steps", 20)):
response = client.messages.create(...)
logger.info(f"[{trace_id}] Step {step+1}: stop_reason={response.stop_reason}, "
f"tokens={response.usage.input_tokens}+{response.usage.output_tokens}")
if response.stop_reason == "tool_use":
for block in response.content:
if block.type == "tool_use":
logger.info(f"[{trace_id}] Tool call: {block.name}({block.input})")
# ... rest of loop
duration = (datetime.utcnow() - start).total_seconds()
logger.info(f"[{trace_id}] Completed in {duration:.1f}s, {step+1} steps")Common Failure Modes
| Failure | Symptom | Fix |
|---|---|---|
| Infinite loops | Agent keeps calling same tool | Add deduplication and step limit |
| Context overflow | Errors after many steps | Summarize old tool results, trim history |
| Tool hallucination | Model invents tool arguments | Add validation in tool executor |
| Ambiguous task | Agent picks wrong subtask | Clarify task upfront or add a planning step |
| Silent failures | Tool returns error, agent ignores it | Use is_error: true flag, add retry logic |
| Cost runaway | Agent loops expensively | Track token usage per run, set budget limits |
Checklist Before Shipping
- Max steps limit is set
- All tool errors are caught and returned as
is_error: true - Token usage is logged per run
- Circuit breakers on all external API calls
- Evals covering happy path + at least 3 failure modes
- Timeout on the whole agent run (not just individual steps)
- Rate limiting if multiple users can trigger agents simultaneously
- Audit log of all tool calls for debugging and compliance
Agents are powerful when they're reliable. Reliability comes from good tool design, explicit failure handling, and observability — not from a smarter model.