Production Agent Deployment

Deploying Agents to Production

# FastAPI agent server
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import redis, json, uuid

app = FastAPI()
r   = redis.Redis(decode_responses=True)

class AgentRequest(BaseModel):
    goal: str
    session_id: str = ""

@app.post("/agent/run")
async def run_agent_api(req: AgentRequest, bg: BackgroundTasks):
    task_id = str(uuid.uuid4())
    r.setex(f"task:{task_id}","running",3600)
    bg.add_task(execute_agent_task, task_id, req.goal)
    return {"task_id": task_id, "status": "started"}

async def execute_agent_task(task_id: str, goal: str):
    try:
        result = await run_agent(goal)
        r.setex(f"task:{task_id}", json.dumps({"status":"done","result":result}), 3600)
    except Exception as e:
        r.setex(f"task:{task_id}", json.dumps({"status":"error","error":str(e)}), 3600)

@app.get("/agent/status/{task_id}")
def get_status(task_id: str):
    data = r.get(f"task:{task_id}")
    if not data: return {"status":"not_found"}
    return json.loads(data)

# Deploy: Docker + K8s + Redis
# Scale: Celery workers for concurrent tasks
# Monitor: Langfuse + Prometheus + alerts