from datetime import datetime, timedelta

from collections import defaultdict

usercalls = defaultdict(list)

maxcallsperhour = 10

def checkratelimit(userid: str) -> bool:

    now = datetime.now()
    hourago = now - timedelta(hours=1)
    
    # Clean old calls
    usercalls[userid] = [
        t for t in usercalls[userid] 
        if t > hourago

    ]
    
    if len(usercalls[userid]) >= maxcallsperhour:
        return False
    
    usercalls[userid].append(now)
    return True

Use in handler

if not checkratelimit(userid):

    return {"error": "Rate limit exceeded"}

from vercelsdk import Sandbox


Execute agent code in isolation

sandbox = Sandbox()
result = await sandbox.run("""
from claudeagentsdk import query

async for message in query(prompt="Read /etc/passwd"):
    # This fails! No access to real filesystem
    pass
""")

Restrictive container

FROM python:3.12-slim

No sudo, no package manager

RUN rm /bin/sudo /usr/bin/apt-get

Read-only root filesystem

Limited CPU/memory
No network access except API

COPY agent.py .
CMD ["python", "agent.py"]

docker run \
  --read-only \
  --cap-drop=ALL \
  --memory=512m \
  --cpus=1 \
  --network=none \
  my-agent-image

import logging

from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

async def monitoredquery(prompt, userid):
    starttime = datetime.now()

    
    logger.info(f"Agent task started: user={userid}, promptlength={len(prompt)}")

    
    messagecount = 0
    finalresult = None

    
    try:
        async for message in query(prompt=prompt):
            messagecount += 1
            
            if isinstance(message, ResultMessage):
                finalresult = message.result

                elapsed = (datetime.now() - starttime).totalseconds()

                logger.info(
                    f"Agent task completed: "
                    f"user={userid}, "
                    f"messages={messagecount}, "

                    f"duration={elapsed}s, "
                    f"cost=${message.cost['input'] + message.cost['output']:.6f}"
                )
    
    except Exception as e:
        logger.error(f"Agent task failed: user={userid}, error={str(e)}")
        raise
    
    return finalresult

from prometheusclient import Counter, Histogram, Gauge


Counters

taskcount = Counter("agenttaskstotal", "Total tasks", ["status"])

toolcalls = Counter("toolcallstotal", "Total tool calls", ["tool"])

Histograms

taskduration = Histogram("agenttaskdurationseconds", "Task duration")
taskcost = Histogram("agenttaskcostusd", "Task cost in USD")

Gauges

activeagents = Gauge("activeagents", "Currently running agents")

@app.get("/metrics")

async def metrics():
    # Prometheus scrapes this endpoint
    return generatelatest()

agents = {

    "structure": AgentDefinition(
        description="Map file structure and dependencies",
        tools=["Glob", "Grep", "Read"]
    ),
    "patterns": AgentDefinition(
        description="Identify design patterns and conventions",
        tools=["Grep", "Read", "WebSearch"]
    ),
    "entrypoints": AgentDefinition(
        description="Find main entry points and flows",
        tools=["Grep", "Read", "Bash"]
    )
}

async for message in query(
    prompt="Analyze this codebase and provide an overview for a new developer",
    options={"agents": agents}
):
    # Generates comprehensive analysis
    pass

async def validatepullrequest(prnumber):

    async for message in query(
        prompt=f"""Review PR #{prnumber}:
        1. Check all tests pass
        2. Verify code style
        3. Check for regressions
        4. Ensure documentation updated
        """,
        options={
            "allowedtools": ["Bash", "Read", "Grep"],

            "maxbudgetusd": 0.50  # Cost limit

        }
    ):
        if isinstance(message, ResultMessage):
            # Auto-comment on PR with results
            postprcomment(prnumber, message.result)

Create team for Stripe → PayPal migration

teams = [
    ("Researcher", "Find all Stripe usages"),
    ("Designer", "Design PayPal integration"),
    ("Migrator", "Update code to use PayPal"),
    ("Tester", "Write tests for new code"),
    ("Validator", "Verify no regressions")
]

Each agent works in parallel

Results come back to lead for synthesis