Deploying Agents to Production

From development to production

Deploying an agent is different from deploying traditional software:

• Stateful interactions: An agent maintains conversation history, tool state, and memory.
• Hard to predict: Agent behavior depends on model outputs, which can change subtly.
• Integration burden: The agent talks to multiple external systems (APIs, databases).
• Cost sensitivity: Every API call and token costs money; waste in production is expensive.

Production deployment requires planning for reliability, observability, and cost control.

Deployment strategies comparison

Different teams choose different deployment patterns based on risk tolerance and rollback speed:

| Strategy | Setup | Rollback Time | Blast Radius | Best For | Risk | |----------|-------|---------------|--------------|----------|------| | Blue-green | 2x infrastructure | Instant (swap LB) | 0% (traffic stays on old version until swap) | Zero-downtime critical services | Higher cost (2x resources) | | Canary | 1x main + monitors | 5–10 min (traffic ramp down) | 5–20% (small traffic %) | Testing risky changes; catch bugs early | Requires good monitoring | | Rolling | 1x, gradual replace | 15–30 min (scale up old version) | 20–100% (ongoing) | Cheaper deployments | Higher risk during rollout | | Shadow | 1x + duplicate traffic | N/A (non-prod) | 0% (mirrored, not live) | Pre-prod validation; no user impact | Doesn't catch prod-only issues (full load, real data) |

Recommendation for agents: Start with canary (5–10% traffic to new version). Agents are non-deterministic, so you need monitors watching error rate, latency, and cost in real-time. If either spikes, automated rollback triggers. Blue-green is overkill for most agents. Rolling is risky because you can't easily rollback mid-deployment if accuracy drops.

Containerization

Docker setup

Package the agent in a Docker container for consistent deployment:

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy agent code
COPY src/ ./src/
COPY config/ ./config/

# Set environment defaults
ENV PYTHONUNBUFFERED=1
ENV LOG_LEVEL=INFO

# Expose port for the API
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
  CMD python -c "import requests; requests.get('http://localhost:8000/health')" || exit 1

# Run the agent server
CMD ["python", "-m", "uvicorn", "src.agent_server:app", "--host", "0.0.0.0", "--port", "8000"]

And a requirements.txt:

anthropic==0.31.0
fastapi==0.104.1
uvicorn==0.24.0
pydantic==2.5.0
python-dotenv==1.0.0
prometheus-client==0.19.0

Configuration via environment

Store configuration in environment variables (not hardcoded):

from pydantic import BaseSettings
import os

class AgentConfig(BaseSettings):
    """Agent configuration from environment variables."""

    # API keys
    anthropic_api_key: str = os.getenv("ANTHROPIC_API_KEY")

    # Model selection
    model_name: str = os.getenv("MODEL_NAME", "claude-3-5-sonnet-20241022")

    # Safety limits
    max_tool_calls: int = int(os.getenv("MAX_TOOL_CALLS", "10"))
    max_tokens: int = int(os.getenv("MAX_TOKENS", "2048"))

    # Logging
    log_level: str = os.getenv("LOG_LEVEL", "INFO")

    # External services
    database_url: str = os.getenv("DATABASE_URL")
    search_api_key: str = os.getenv("SEARCH_API_KEY")

    class Config:
        env_file = ".env"
        case_sensitive = False

config = AgentConfig()

Alternatively, use a multi-stage Docker build to minimize image size:

# Dockerfile: multi-stage build
FROM python:3.11 as builder

WORKDIR /build
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

FROM python:3.11-slim

WORKDIR /app

# Copy only the installed packages (not the entire pip cache)
COPY --from=builder /root/.local /root/.local
ENV PATH=/root/.local/bin:$PATH

# Copy agent code
COPY src/ ./src/
COPY config/ ./config/

ENV PYTHONUNBUFFERED=1
ENV LOG_LEVEL=INFO
EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
  CMD python -c "import requests; requests.get('http://localhost:8000/health')" || exit 1

CMD ["python", "-m", "uvicorn", "src.agent_server:app", "--host", "0.0.0.0", "--port", "8000"]

This reduces image size from ~800MB to ~400MB, speeding up deployments.

Deploy with environment variables:

docker run -d \
  -e ANTHROPIC_API_KEY="sk-..." \
  -e DATABASE_URL="postgresql://user:pass@db:5432/agent_db" \
  -e MAX_TOOL_CALLS="5" \
  -p 8000:8000 \
  my-agent:latest

Health checks and readiness

Liveness probe

Detect if the agent process is stuck or dead:

from fastapi import FastAPI, HTTPException
import asyncio

app = FastAPI()

@app.get("/health")
async def health_check():
    """Liveness probe: is the service responding?"""
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat()
    }

@app.get("/ready")
async def readiness_check():
    """Readiness probe: can the service handle traffic?"""
    try:
        # Check critical dependencies
        if not model_client.models.list():
            raise Exception("Cannot reach Anthropic API")

        if not db.ping():
            raise Exception("Database connection failed")

        return {"status": "ready"}

    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Service not ready: {e}")

@app.post("/agent/chat")
async def agent_chat(request: ChatRequest):
    """Handle chat requests."""
    try:
        result = agent.run(request.message)
        return {"response": result, "status": "success"}
    except Exception as e:
        logger.error(f"Agent error: {e}")
        raise HTTPException(status_code=500, detail="Agent failed")

Use the health checks in Kubernetes:

# kubernetes-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-service
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: agent
        image: my-agent:latest
        ports:
        - containerPort: 8000

        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 30

        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 10

Observability: logging and metrics

Structured logging

Log all agent actions in a searchable format:

import json
import logging
from pythonjsonlogger import jsonlogger

# Configure JSON logging for easy searching
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)
logger = logging.getLogger()
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

def log_agent_execution(user_id: str, input_text: str, output: str, tool_calls: int, duration_ms: float):
    """Log structured data about an agent execution."""
    logger.info("agent_execution", extra={
        "user_id": user_id,
        "input_length": len(input_text),
        "output_length": len(output),
        "tool_calls": tool_calls,
        "duration_ms": duration_ms,
        "timestamp": datetime.now().isoformat(),
    })

# Usage:
start = time.time()
result = agent.run(user_message)
duration = (time.time() - start) * 1000
log_agent_execution(user_id, user_message, result, tool_count, duration)

Metrics collection

Track performance metrics for alerting:

from prometheus_client import Counter, Histogram, Gauge

# Metrics
agent_requests = Counter(
    'agent_requests_total',
    'Total agent requests',
    ['status']
)

agent_latency = Histogram(
    'agent_latency_seconds',
    'Agent execution latency',
    buckets=(0.5, 1.0, 2.0, 5.0, 10.0)
)

agent_errors = Counter(
    'agent_errors_total',
    'Total agent errors',
    ['error_type']
)

tool_calls = Counter(
    'tool_calls_total',
    'Total tool calls',
    ['tool_name', 'status']
)

@app.post("/agent/chat")
async def agent_chat(request: ChatRequest):
    start = time.time()
    try:
        result = agent.run(request.message)
        agent_requests.labels(status='success').inc()
        agent_latency.observe(time.time() - start)
        return {"response": result}
    except Exception as e:
        agent_requests.labels(status='error').inc()
        agent_errors.labels(error_type=type(e).__name__).inc()
        raise

@app.get("/metrics")
async def metrics():
    """Expose metrics for Prometheus scraping."""
    from prometheus_client import generate_latest
    return generate_latest()

Rate limiting and cost control

Per-user quota

Limit agent usage per user to control costs:

from datetime import datetime, timedelta
import redis

class RateLimiter:
    """Rate limit agent calls per user."""

    def __init__(self, redis_client):
        self.redis = redis_client
        self.default_quota = 100  # Calls per day

    def is_allowed(self, user_id: str, quota: int = None) -> bool:
        """Check if user can make another call."""
        quota = quota or self.default_quota
        key = f"agent_calls:{user_id}:{datetime.now().date()}"

        current = int(self.redis.get(key) or 0)
        if current >= quota:
            return False

        self.redis.incr(key)
        self.redis.expire(key, 86400)  # Reset daily
        return True

limiter = RateLimiter(redis_client)

@app.post("/agent/chat")
async def agent_chat(request: ChatRequest, user_id: str):
    if not limiter.is_allowed(user_id):
        raise HTTPException(status_code=429, detail="Rate limit exceeded")

    result = agent.run(request.message)
    return {"response": result}

Token budgets

Track and limit token spend per request:

def run_agent_with_token_budget(
    user_message: str,
    max_tokens: int = 2048,
    max_total_tokens: int = 5000
) -> str:
    """Run agent but stop if token usage exceeds budget."""

    total_tokens = 0
    context = []

    while total_tokens < max_total_tokens:
        response = model.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=max_tokens,
            messages=context + [{"role": "user", "content": user_message}]
        )

        total_tokens += response.usage.input_tokens + response.usage.output_tokens

        if total_tokens > max_total_tokens:
            logger.warning(f"Token budget exceeded: {total_tokens} > {max_total_tokens}")
            return "Agent hit token limit. Request was too complex."

        context.append({"role": "assistant", "content": response.content[0].text})

        if response.stop_reason == "end_turn":
            return response.content[0].text

    return "Agent exhausted token budget."

Graceful degradation

Fallback responses

When the agent fails, provide a useful fallback:

class GracefulAgent:
    """Agent with fallback for failures."""

    def run(self, user_message: str) -> dict:
        """Execute agent with multiple fallback strategies."""

        # Strategy 1: Try the full agent
        try:
            result = self.agent.run(user_message)
            return {"response": result, "fallback_level": 0}
        except TimeoutError:
            logger.warning("Agent timed out, using fallback 1")

        # Strategy 2: Try a simpler model without tools
        try:
            response = model.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=512,
                messages=[{"role": "user", "content": user_message}]
            )
            return {"response": response.content[0].text, "fallback_level": 1}
        except Exception:
            logger.warning("Fallback 1 failed, using fallback 2")

        # Strategy 3: Return a canned response
        return {
            "response": "I'm experiencing technical difficulties. Please try again in a moment.",
            "fallback_level": 2
        }

@app.post("/agent/chat")
async def agent_chat(request: ChatRequest):
    result = GracefulAgent().run(request.message)

    # Flag non-standard responses for investigation
    if result["fallback_level"] > 0:
        logger.error(f"Agent used fallback level {result['fallback_level']}")

    return result

Circuit breaker

If a service (API, database) is consistently failing, stop calling it:

from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing; block calls
    HALF_OPEN = "half_open"  # Testing if service recovered

class CircuitBreaker:
    """Prevent cascading failures by stopping calls to a failing service."""

    def __init__(self, failure_threshold: int = 5, timeout_seconds: int = 60):
        self.state = CircuitState.CLOSED
        self.failures = 0
        self.failure_threshold = failure_threshold
        self.timeout_seconds = timeout_seconds
        self.last_failure_time = None

    def call(self, func, *args, **kwargs):
        """Execute a function, tracking failures and opening circuit if needed."""

        if self.state == CircuitState.OPEN:
            if (datetime.now() - self.last_failure_time).seconds > self.timeout_seconds:
                # Try half-open: test if service recovered
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is OPEN; service unavailable")

        try:
            result = func(*args, **kwargs)

            # Success: reset state
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
            self.failures = 0
            return result

        except Exception as e:
            self.failures += 1
            self.last_failure_time = datetime.now()

            if self.failures >= self.failure_threshold:
                self.state = CircuitState.OPEN
                logger.error(f"Circuit breaker OPEN after {self.failures} failures")

            raise

# Usage:
breaker = CircuitBreaker(failure_threshold=3)

def call_external_api(*args, **kwargs):
    """Wrapper that uses circuit breaker."""
    return breaker.call(api.call, *args, **kwargs)

Case Study: Canary deployment catches a silent failure

A company deployed a new version of their agent to production using the canary strategy (5% traffic). The new version had a higher accuracy (94% vs. 91%), and the team was excited to roll it out. But the canary monitoring caught something subtle:

• Accuracy was indeed 94% on the new version.
• But latency increased from 1.2s to 3.8s — a 3x slowdown.
• Cost per request jumped from $0.008 to $0.025 — a 3x increase.

Why? The new version used a more sophisticated reasoning approach that made more tool calls and longer prompts. In tests (small batch, no concurrent load), this went unnoticed. But in production (10,000 concurrent users), the extra latency cascaded: users waited longer, timeouts increased, the service became unstable.

The canary caught it before it affected all users. The team rolled back, optimized the reasoning algorithm to be more efficient, and re-deployed. Lesson: Latency and cost matter in production more than small accuracy gains. Canary deployment with automated monitors is essential for non-deterministic systems.

Incident response

Alerting on anomalies

Configure alerts for unusual behavior:

import smtplib
from email.mime.text import MIMEText

def alert_ops(subject: str, message: str):
    """Send alert to operations team."""
    msg = MIMEText(message)
    msg['Subject'] = subject
    msg['From'] = "[email protected]"
    msg['To'] = "[email protected]"

    with smtplib.SMTP_SSL('smtp.gmail.com', 465) as server:
        server.login(os.getenv("ALERT_EMAIL"), os.getenv("ALERT_PASSWORD"))
        server.send_message(msg)

# Example: Alert if error rate is high
def check_error_rate():
    """Check error rate and alert if too high."""
    total_requests = agent_requests.labels(status='success')._value.get() + \
                     agent_requests.labels(status='error')._value.get()

    if total_requests == 0:
        return

    error_rate = agent_requests.labels(status='error')._value.get() / total_requests

    if error_rate > 0.1:  # More than 10% errors
        alert_ops(
            subject="Agent Error Rate High",
            message=f"Error rate is {error_rate:.1%} ({agent_requests.labels(status='error')._value.get()} errors in last {total_requests} requests)"
        )

Edge case: The "silent degradation" scenario

Production monitoring can miss failures that don't raise exceptions. Consider:

# Silent degradation example:
# Agent used to answer factual questions with 90% accuracy.
# A model update changed the reasoning slightly.
# New behavior: Agent answers correctly but now uses 5x more tokens (cost skyrockets).
# Accuracy monitoring shows: 89% (slight drop, within normal variance).
# Cost monitoring shows: $0.04/request (was $0.008) — a 5x increase.
# Error rate: 0% (no errors; system "works").
# Result: System is "working" but bleeding money unnoticed.

To detect silent degradation:

1. Monitor derived metrics, not just primary ones:

   • Don't just track "accuracy" — track "cost per accurate answer" (accuracy / cost).
   • Don't just track "latency" — track "latency under load" (p99 latency during peak traffic).

2. Set up cost anomaly detection: If cost per request increases >20% over a week, alert automatically.

3. Regular audits: Once per sprint, manually sample 50 production interactions. Does the agent's behavior match your expectations?

# Example: Cost per accurate answer metric
def cost_per_accurate_answer(cost_per_request, accuracy):
    """Lower is better."""
    if accuracy == 0:
        return float('inf')
    return cost_per_request / accuracy

# Track this as a custom metric
cost_per_answer = cost_per_accurate_answer(0.04, 0.89)  # = $0.045 per correct answer
# Alert if this goes above a threshold (e.g., $0.01)

Runbook for common issues

Document recovery procedures:

## Agent High Latency

Symptoms:
- /metrics shows agent_latency_seconds > 5s for 5+ minutes
- User complaints of slow responses

Diagnosis:
1. Check external API status: curl https://api.example.com/health
2. Check database latency: psql -c "SELECT version();"
3. Check logs for errors: tail -100 /var/log/agent.log

Recovery:
- If API is slow: restart agent container, it will skip tool calls
- If database is slow: contact DBA; agent will degrade to non-database mode
- If neither, increase container timeout and redeploy

Common mistake

Deploying to production and then assuming it's done. Production is a continuous process: monitor metrics, collect feedback, iterate on reliability. The first few days in production will reveal issues that tests missed. Have a plan to respond quickly, revert if needed, and learn from each incident.