How to Build Scalable Python Backends for AI Agents and Automation Workflows

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
 
from app.api.routes import agents, tasks
from app.core.cache import redis_client
from app.core.database import engine
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await redis_client.connect()
    yield
    # Shutdown
    await redis_client.disconnect()
 
app = FastAPI(
    title="AI Agent Backend",
    version="1.0.0",
    lifespan=lifespan
)
 
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
app.include_router(agents.router, prefix="/api/v1/agents", tags=["agents"])
app.include_router(tasks.router, prefix="/api/v1/tasks", tags=["tasks"])
 
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
 
class Settings(BaseSettings):
    # API Keys
    OPENAI_API_KEY: str
    MISTRAL_API_KEY: str
    
    # Database
    DATABASE_URL: str
    
    # Redis
    REDIS_URL: str
    
    # App Settings
    DEBUG: bool = False
    API_V1_PREFIX: str = "/api/v1"
    
    class Config:
        env_file = ".env"
 
@lru_cache()
def get_settings():
    return Settings()

# ❌ BAD: Blocks the entire event loop
def generate_response(prompt: str):
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# ✅ GOOD: Non-blocking
async def generate_response(prompt: str):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={
                "model": "gpt-4",
                "messages": [{"role": "user", "content": prompt}]
            }
        )
    return response.json()["choices"][0]["message"]["content"]

# app/core/ai_client.py
import httpx
from typing import List, Dict
from app.config import get_settings
 
class AIClient:
    def __init__(self):
        self.settings = get_settings()
        self.client = httpx.AsyncClient(timeout=30.0)
    
    async def chat_completion(
        self,
        messages: List[Dict[str, str]],
        model: str = "gpt-4",
        temperature: float = 0.7
    ) -> str:
        """Generate AI response"""
        try:
            response = await self.client.post(
                "https://api.openai.com/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {self.settings.OPENAI_API_KEY}"
                },
                json={
                    "model": model,
                    "messages": messages,
                    "temperature": temperature
                }
            )
            response.raise_for_status()
            return response.json()["choices"][0]["message"]["content"]
        except httpx.HTTPError as e:
            # Log error
            raise Exception(f"AI API error: {str(e)}")
    
    async def close(self):
        await self.client.aclose()
 
# Dependency injection
async def get_ai_client():
    client = AIClient()
    try:
        yield client
    finally:
        await client.close()

from fastapi import BackgroundTasks
 
@app.post("/api/v1/agents/process")
async def process_agent_task(
    task_data: TaskSchema,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(run_agent_workflow, task_data)
    return {"status": "processing", "task_id": task_data.id}
 
async def run_agent_workflow(task_data: TaskSchema):
    # Long-running AI task
    result = await ai_client.chat_completion(task_data.messages)
    await save_result(result)

# app/core/celery_app.py
from celery import Celery
 
celery_app = Celery(
    "ai_backend",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0"
)
 
@celery_app.task
def process_ai_task(task_id: str, prompt: str):
    # This runs in a separate worker process
    result = call_ai_api(prompt)
    save_to_db(task_id, result)
    return result

# In your API endpoint
@app.post("/api/v1/agents/process")
async def process_agent_task(task_data: TaskSchema):
    task = process_ai_task.delay(task_data.id, task_data.prompt)
    return {"status": "queued", "task_id": task.id}

# app/core/cache.py
import redis.asyncio as redis
import json
from typing import Optional
 
class RedisCache:
    def __init__(self, url: str):
        self.redis = redis.from_url(url)
    
    async def get(self, key: str) -> Optional[dict]:
        data = await self.redis.get(key)
        return json.loads(data) if data else None
    
    async def set(self, key: str, value: dict, ttl: int = 3600):
        await self.redis.setex(
            key,
            ttl,
            json.dumps(value)
        )
    
    async def delete(self, key: str):
        await self.redis.delete(key)
 
# Usage in service
async def get_agent_response(prompt: str, cache: RedisCache):
    cache_key = f"ai:response:{hash(prompt)}"
    
    # Check cache first
    cached = await cache.get(cache_key)
    if cached:
        return cached
    
    # Generate new response
    response = await ai_client.chat_completion([
        {"role": "user", "content": prompt}
    ])
    
    # Cache it
    await cache.set(cache_key, {"response": response}, ttl=3600)
    
    return {"response": response}

from tenacity import retry, stop_after_attempt, wait_exponential
 
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def call_ai_with_retry(prompt: str):
    try:
        return await ai_client.chat_completion([
            {"role": "user", "content": prompt}
        ])
    except Exception as e:
        logger.error(f"AI API call failed: {str(e)}")
        raise

import logging
from pythonjsonlogger import jsonlogger
 
# Structured logging
logger = logging.getLogger()
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)
 
# Usage
logger.info("AI request", extra={
    "user_id": user.id,
    "prompt_length": len(prompt),
    "model": "gpt-4",
    "latency_ms": latency
})

# app/models/agent.py
from sqlalchemy import Column, String, JSON, DateTime, Enum
from sqlalchemy.dialects.postgresql import UUID
import uuid
from datetime import datetime
 
from app.core.database import Base
 
class AgentTask(Base):
    __tablename__ = "agent_tasks"
    
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    user_id = Column(UUID(as_uuid=True), nullable=False)
    prompt = Column(String, nullable=False)
    response = Column(String)
    metadata = Column(JSON)
    status = Column(Enum("pending", "processing", "completed", "failed"))
    created_at = Column(DateTime, default=datetime.utcnow)
    completed_at = Column(DateTime)

# app/api/routes/agents.py
from fastapi import APIRouter, Depends, HTTPException
from app.schemas.agent import AgentRequest, AgentResponse
from app.services.agent_service import AgentService
from app.core.ai_client import get_ai_client
 
router = APIRouter()
 
@router.post("/chat", response_model=AgentResponse)
async def chat_with_agent(
    request: AgentRequest,
    ai_client: AIClient = Depends(get_ai_client)
):
    """Process agent chat request"""
    try:
        response = await ai_client.chat_completion(
            messages=request.messages,
            model=request.model
        )
        return AgentResponse(
            response=response,
            model=request.model,
            tokens_used=len(response.split())
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# tests/test_agents.py
import pytest
from httpx import AsyncClient
from app.main import app
 
@pytest.mark.asyncio
async def test_chat_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/agents/chat",
            json={
                "messages": [{"role": "user", "content": "Hello"}],
                "model": "gpt-4"
            }
        )
    assert response.status_code == 200
    assert "response" in response.json()

# Dockerfile
FROM python:3.11-slim
 
WORKDIR /app
 
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY . .
 
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'
 
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/aidb
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
  
  db:
    image: postgres:15
    environment:
      POSTGRES_DB: aidb
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
  
  redis:
    image: redis:7-alpine