I’ve been running AI agents in production for over a year now. Most tutorials skip the hard parts - state, rate limits, and what happens when your agent calls a tool that times out. Here’s how I build them with FastAPI and the OpenAI SDK, the way I’d want it documented when I’m paged at 2 a.m.
Start with a minimal FastAPI app that accepts a prompt, calls OpenAI, and returns a response. No tools, no state - just the core loop. This is your foundation.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import openai
import os
app = FastAPI()
client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
class AgentRequest(BaseModel):
prompt: str
model: str = "gpt-4o"
class AgentResponse(BaseModel):
response: str
@app.post("/agent", response_model=AgentResponse)
async def run_agent(request: AgentRequest):
try:
completion = client.chat.completions.create(
model=request.model,
messages=[{"role": "user", "content": request.prompt}]
)
return AgentResponse(response=completion.choices[0].message.content)
except openai.RateLimitError:
raise HTTPException(status_code=429, detail="Rate limit exceeded")
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
This works for simple Q&A. But real agents need to do more than chat - they need to act.
Tool use turns your agent from a talker into a doer. I define functions the agent can call - like querying a database or hitting an internal API - and let the OpenAI SDK handle the function calling loop.
from typing import Optional
import httpx
import os
async def get_user_info(user_id: str) -> dict:
async with httpx.AsyncClient() as client:
resp = await client.get(f"{os.getenv('INTERNAL_API_URL')}/users/{user_id}")
resp.raise_for_status()
return resp.json()
from openai import OpenAI
from typing import List, Dict, Any
import json
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
TOOLS = [
{
"type": "function",
"function": {
"name": "get_user_info",
"description": "Fetch user details by ID",
"parameters": {
"type": "object",
"properties": {
"user_id": {"type": "string", "description": "The user ID"}
},
"required": ["user_id"]
}
}
}
]
async def run_agent_with_tools(prompt: str, model: str = "gpt-4o") -> str:
messages = [{"role": "user", "content": prompt}]
while True:
completion = client.chat.completions.create(
model=model,
messages=messages,
tools=TOOLS,
tool_choice="auto"
)
message = completion.choices[0].message
if message.tool_calls:
messages.append(message)
for tool_call in message.tool_calls:
if tool_call.function.name == "get_user_info":
args = json.loads(tool_call.function.arguments)
result = await get_user_info(args["user_id"])
messages.append({
"tool_call_id": tool_call.id,
"role": "tool",
"content": json.dumps(result)
})
continue
else:
return message.content
I’ve seen teams try to manage this loop manually with state machines. Don’t. Let the SDK handle the tool call/response cycle - it’s less buggy and easier to debug.
Production agents aren’t stateless. They need memory across turns - chat history, user preferences, cached tool results. I use Redis for fast state and Pydantic models to keep it typed.
import redis.asyncio as redis
from pydantic import BaseModel
from typing import Optional, Dict, Any
import json
redis_client = redis.from_url(os.getenv("REDIS_URL"), decode_responses=True)
class AgentState(BaseModel):
session_id: str
chat_history: List[Dict[str, str]] = []
user_preferences: Dict[str, Any] = {}
last_tool_result: Optional[Dict] = None
async def get_state(session_id: str) -> AgentState:
data = await redis_client.get(f"agent_state:{session_id}")
if data:
return AgentState.model_validate_json(data)
return AgentState(session_id=session_id)
async def save_state(state: AgentState):
await redis_client.set(
f"agent_state:{state.session_id}",
state.model_dump_json(),
ex=3600 # 1 hour TTL
)
Each request loads state, runs the agent loop (which may mutate state via tools), then saves it back. I’ve had agents corrupt state by writing concurrently - use Redis transactions or a queue if you need strong consistency.
Containerize it. Cloud Run scales to zero, which saves money when traffic is spiky. But cold starts hurt - keep your image small and avoid heavy imports at module level.
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV PORT=8080
EXPOSE 8080
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]
I deploy with:
gcloud builds submit --tag gcr.io/my-project/ai-agent
gcloud run deploy ai-agent --image gcr.io/my-project/ai-agent --platform managed
Watch for:
--secret in Cloud Build or Cloud Run env vars.
Rate limits are the most frequent pager. I’ve seen agents get stuck in retry loops because they didn’t back off. Token limits sneak up when chat history grows. Context overflow makes agents forget instructions.
Here’s how I defend against them:
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import time
from collections import defaultdict
request_counts = defaultdict(list)
class RateLimitMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
client_ip = request.client.host
now = time.time()
request_counts[client_ip] = [t for t in request_counts[client_ip] if now - t < 60]
if len(request_counts[client_ip]) >= 10: # 10 req/min
return Response("Rate limit exceeded", status_code=429)
request_counts[client_ip].append(now)
return await call_next(request)
For token limits, I truncate history:
def truncate_history(messages, max_tokens=3000):
total_chars = sum(len(m["content"]) for m in messages)
if total_chars > max_tokens * 4:
return [messages[0]] + messages[-(max_tokens//4):] # Simplified
return messages
I log token usage per request:
usage = completion.usage
logger.info(f"Token usage: {usage.total_tokens} (prompt: {usage.prompt_tokens}, completion: {usage.completion_tokens})")
If you see completion tokens near the limit, your agent is looping or over-explaining.
I track three things: success rate, latency, and tool accuracy. Success rate means the agent completed the user’s goal without human intervention. I log every turn and use a simple evaluator LLM to judge outcomes.
from prometheus_client import Counter, Histogram
import time
AGENT_REQUESTS = Counter('agent_requests_total', 'Total agent requests', ['status'])
AGENT_LATENCY = Histogram('agent_latency_seconds', 'Time spent processing agent requests')
TOOL_USAGE = Counter('agent_tool_usage_total', 'Tool usage count', ['tool_name'])
async def run_agent_evaluated(prompt: str, session_id: str):
start = time.time()
state = await get_state(session_id)
try:
response = await run_agent_with_tools(prompt)
success = "error" not in response.lower()
AGENT_REQUESTS.labels(status="success" if success else "failure").inc()
logger.info({
"session_id": session_id,
"prompt": prompt,
"response": response,
"state": state.model_dump(),
"success": success
})
return response
finally:
AGENT_LATENCY.observe(time.time() - start)
I’ve used this to catch agents that call tools unnecessarily - wasting money and latency. If your tool usage counter is high but success rate low, your agent is confused about when to act.
A FastAPI endpoint that calls OpenAI chat.completions.create with a user prompt and returns the message content. Add error handling for rate limits and you’ve got a baseline.
Limit the number of tool call iterations per request (e.g., max 5 turns). The OpenAI SDK doesn’t do this by default - wrap your agent loop in a counter and break if exceeded.
For production agents needing fine-grained control over tool calls, state, and logging, use the OpenAI SDK directly. LangChain adds abstraction that obscures failure modes and makes debugging harder.
At $0.005 per 1K tokens for GPT-4o, a typical agent turn (500 prompt + 150 completion tokens) costs ~$0.003. Tool calls add latency but no extra token cost - watch for iteration bloat.