@app.post("/analyze")
async def analyze(request: AnalyzeRequest):
    result = agent.run(request.query)
    return {"result": result}

1@app.post("/analyze")
2async def start_analysis(request: AnalyzeRequest):
3    job_id = create_job(request)
4    queue.enqueue(run_agent_job, job_id)
5    return {"job_id": job_id}
6 
7@app.get("/jobs/{job_id}")
8async def get_job(job_id: str):
9    return get_job_status(job_id)

1@app.post("/chat/stream")
2async def stream_chat(request: ChatRequest):
3    async def generate():
4        async for chunk in agent.stream(request.message):
5            yield f"data: {json.dumps(chunk)}\n\n"
6 
7    return StreamingResponse(generate(), media_type="text/event-stream")

1Level 1: Simple Chat
2└── User asks, agent responds
3 
4Level 2: Show Thinking
5└── Display reasoning steps
6 
7Level 3: Tool Visibility
8└── Show which tools are used
9 
10Level 4: Full Control
11└── User guides decisions

1function AgentStatus({ jobId }) {
2  const [status, setStatus] = useState('starting');
3  const [steps, setSteps] = useState([]);
4 
5  useEffect(() => {
6    const events = new EventSource(`/jobs/${jobId}/stream`);
7 
8    events.onmessage = (e) => {
9      const data = JSON.parse(e.data);
10      setStatus(data.status);
11      setSteps(data.steps);
12    };
13 
14    return () => events.close();
15  }, [jobId]);
16 
17  return (
18    <div>
19      <StatusBadge status={status} />
20      <StepsList steps={steps} />
21      <ProgressBar progress={steps.length / totalSteps} />
22    </div>
23  );
24}

1FROM python:3.11-slim
2 
3WORKDIR /app
4COPY requirements.txt .
5RUN pip install --no-cache-dir -r requirements.txt
6COPY . .
7 
8CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

1# docker-compose.yml
2version: '3.8'
3 
4services:
5  api:
6    build: .
7    ports:
8      - "8000:8000"
9    environment:
10      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
11      - REDIS_URL=redis://redis:6379
12    depends_on:
13      - redis
14 
15  worker:
16    build: .
17    command: celery -A tasks worker --loglevel=info
18    environment:
19      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
20      - REDIS_URL=redis://redis:6379
21 
22  redis:
23    image: redis:7-alpine

1# AWS Lambda handler
2def handler(event, context):
3    body = json.loads(event['body'])
4    result = agent.run(body['query'])
5 
6    return {
7        'statusCode': 200,
8        'body': json.dumps({'result': result})
9    }

1apiVersion: apps/v1
2kind: Deployment
3metadata:
4  name: agent-service
5spec:
6  replicas: 3
7  template:
8    spec:
9      containers:
10      - name: agent
11        image: your-registry/agent:latest
12        resources:
13          requests:
14            memory: "512Mi"
15            cpu: "500m"
16          limits:
17            memory: "1Gi"
18            cpu: "1000m"
19        env:
20        - name: ANTHROPIC_API_KEY
21          valueFrom:
22            secretKeyRef:
23              name: api-keys
24              key: anthropic

1import structlog
2 
3logger = structlog.get_logger()
4 
5def run_with_logging(user_id: str, task: str):
6    log = logger.bind(user_id=user_id, task_id=str(uuid4()))
7    log.info("agent_start", task=task)
8 
9    try:
10        start = time.time()
11        result = agent.run(task)
12        duration = time.time() - start
13 
14        log.info("agent_done",
15                 duration=duration,
16                 tokens=result.tokens,
17                 tools=result.tools)
18        return result
19 
20    except Exception as e:
21        log.error("agent_failed", error=str(e))
22        raise

1# Prometheus alert rules
2groups:
3- name: agent-alerts
4  rules:
5  - alert: HighLatency
6    expr: histogram_quantile(0.95, agent_duration_seconds) > 30
7    for: 5m
8    labels:
9      severity: warning
10 
11  - alert: HighToolFailures
12    expr: rate(tool_failures[5m]) / rate(tool_calls[5m]) > 0.05
13    for: 5m
14    labels:
15      severity: critical

1class BudgetedAgent:
2    def __init__(self, max_tokens: int = 100000):
3        self.max_tokens = max_tokens
4        self.tokens_used = 0
5 
6    def run(self, task: str) -> str:
7        if self.tokens_used >= self.max_tokens:
8            raise BudgetExceededError()
9 
10        response = self._call_api(task)
11        self.tokens_used += response.usage.total_tokens
12        return response

1import hashlib
2 
3class CachedAgent:
4    def __init__(self):
5        self.cache = {}
6 
7    def run(self, task: str) -> str:
8        key = hashlib.md5(task.encode()).hexdigest()
9 
10        if key in self.cache:
11            return self.cache[key]
12 
13        result = self._run(task)
14        self.cache[key] = result
15        return result

1TIERS = {
2    "free": {"tokens_day": 10000, "tasks_day": 10},
3    "pro": {"tokens_day": 100000, "tasks_day": 100},
4    "enterprise": {"tokens_day": None, "tasks_day": None}
5}
6 
7def check_usage(user_id: str, tier: str) -> bool:
8    usage = get_daily_usage(user_id)
9    limits = TIERS[tier]
10 
11    if limits["tokens_day"] and usage.tokens > limits["tokens_day"]:
12        return False
13    if limits["tasks_day"] and usage.tasks > limits["tasks_day"]:
14        return False
15    return True

1from pydantic import BaseModel, validator
2 
3class AgentRequest(BaseModel):
4    task: str
5    max_tokens: int = 4096
6 
7    @validator('task')
8    def validate_task(cls, v):
9        if len(v) > 10000:
10            raise ValueError("Task too long")
11        if contains_pii(v):
12            raise ValueError("Remove personal information")
13        return v

1def filter_output(response: str) -> str:
2    response = remove_system_paths(response)
3    response = redact_secrets(response)
4 
5    if contains_harmful_content(response):
6        return "I cannot provide that information."
7 
8    return response

1SENSITIVE_ACTIONS = ["delete_file", "send_email", "make_payment"]
2 
3def execute_with_approval(tool: str, params: dict, user_id: str) -> str:
4    if tool in SENSITIVE_ACTIONS:
5        request_id = create_approval_request(user_id, tool, params)
6        approved = wait_for_approval(request_id, timeout=300)
7 
8        if not approved:
9            return "Action not approved."
10 
11    return execute_tool(tool, params)

1from slowapi import Limiter
2 
3limiter = Limiter(key_func=get_remote_address)
4 
5@app.post("/agent")
6@limiter.limit("10/minute")
7async def run_agent(request: Request):
8    # ...

1Pay per task:
2- Free: 10 tasks/month
3- Basic: $0.10/task (up to 1000)
4- Pro: $0.05/task (unlimited)
5 
6Pay per token:
7- $0.01 per 1000 tokens

1Free Tier:
2- 100 tasks/month
3- Basic tools
4- Community support
5 
6Pro ($29/month):
7- 1000 tasks/month
8- All tools
9- Priority support
10 
11Enterprise (Custom):
12- Unlimited
13- Custom tools
14- SLA guarantee

1# content_agent.py
2class ContentAgent:
3    def __init__(self, brand_voice: str = None):
4        self.client = Anthropic()
5        self.brand_voice = brand_voice
6 
7    async def generate_blog(self, topic: str, keywords: list) -> str:
8        # Research topic
9        research = await self._research(topic)
10 
11        # Generate outline
12        outline = await self._create_outline(topic, research)
13 
14        # Write content
15        content = await self._write_content(outline, keywords)
16 
17        # Apply brand voice
18        if self.brand_voice:
19            content = await self._apply_voice(content)
20 
21        return content