swarms/docs/deployment_solutions/fastapi_agent_api.md

FastAPI Agent API

This guide shows you how to deploy your Swarms agents as REST APIs using FastAPI and Uvicorn. This is the fastest way to expose your agents via HTTP endpoints.

Overview

FastAPI is a modern, fast web framework for building APIs with Python. Combined with Uvicorn (ASGI server), it provides excellent performance and automatic API documentation.

Benefits:

Feature	Description
Fast	Built on Starlette and Pydantic
Auto-docs	Automatic OpenAPI/Swagger documentation
Type-safe	Full type hints and validation
Easy	Minimal boilerplate code
Monitoring	Built-in logging and metrics

Quick Start

1. Install Dependencies

pip install fastapi uvicorn swarms

2. Create Your Agent API

Create a file called agent_api.py:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from swarms import Agent
import uvicorn
from typing import Optional, Dict, Any

# Initialize FastAPI app
app = FastAPI(
    title="Swarms Agent API",
    description="REST API for Swarms agents",
    version="1.0.0"
)

# Pydantic models for request/response
class AgentRequest(BaseModel):
    """Request model for agent tasks"""
    task: str
    agent_name: Optional[str] = "default"
    max_loops: Optional[int] = 1
    temperature: Optional[float] = None

class AgentResponse(BaseModel):
    """Response model for agent tasks"""
    success: bool
    result: str
    agent_name: str
    task: str
    execution_time: Optional[float] = None

# Initialize your agent (you can customize this)
def create_agent(agent_name: str = "default") -> Agent:
    """Create and return a configured agent"""
    return Agent(
        agent_name=agent_name,
        agent_description="Versatile AI agent for various tasks",
        system_prompt="""You are a helpful AI assistant that can handle a wide variety of tasks.
        You provide clear, accurate, and helpful responses while maintaining a professional tone.
        Always strive to be thorough and accurate in your responses.""",
        model_name="claude-sonnet-4-20250514",
        dynamic_temperature_enabled=True,
        max_loops=1,
        dynamic_context_window=True,
    )

# API endpoints
@app.get("/")
async def root():
    """Health check endpoint"""
    return {"message": "Swarms Agent API is running!", "status": "healthy"}

@app.get("/health")
async def health_check():
    """Detailed health check"""
    return {
        "status": "healthy",
        "service": "Swarms Agent API",
        "version": "1.0.0"
    }

@app.post("/agent/run", response_model=AgentResponse)
async def run_agent(request: AgentRequest):
    """Run an agent with the specified task"""
    try:
        import time
        start_time = time.time()
        
        # Create agent instance
        agent = create_agent(request.agent_name)
        
        # Run the agent
        result = agent.run(
            task=request.task,
            max_loops=request.max_loops
        )
        
        execution_time = time.time() - start_time
        
        return AgentResponse(
            success=True,
            result=str(result),
            agent_name=request.agent_name,
            task=request.task,
            execution_time=execution_time
        )
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent execution failed: {str(e)}")

@app.post("/agent/chat")
async def chat_with_agent(request: AgentRequest):
    """Chat with an agent (conversational mode)"""
    try:
        agent = create_agent(request.agent_name)
        
        # For chat, you might want to maintain conversation history
        # This is a simple implementation
        result = agent.run(
            task=request.task,
            max_loops=request.max_loops
        )
        
        return {
            "success": True,
            "response": str(result),
            "agent_name": request.agent_name
        }
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Chat failed: {str(e)}")

@app.get("/agents/available")
async def list_available_agents():
    """List available agent configurations"""
    return {
        "agents": [
            {
                "name": "default",
                "description": "Versatile AI agent for various tasks",
                "model": "claude-sonnet-4-20250514"
            },
            {
                "name": "quantitative-trading",
                "description": "Advanced quantitative trading and algorithmic analysis agent",
                "model": "claude-sonnet-4-20250514"
            }
        ]
    }

# Custom agent endpoint example
@app.post("/agent/quantitative-trading")
async def run_quantitative_trading_agent(request: AgentRequest):
    """Run the quantitative trading agent specifically"""
    try:
        # Create specialized quantitative trading agent
        agent = Agent(
            agent_name="Quantitative-Trading-Agent",
            agent_description="Advanced quantitative trading and algorithmic analysis agent",
            system_prompt="""You are an expert quantitative trading agent with deep expertise in:
            - Algorithmic trading strategies and implementation
            - Statistical arbitrage and market making
            - Risk management and portfolio optimization
            - High-frequency trading systems
            - Market microstructure analysis
            - Quantitative research methodologies
            - Financial mathematics and stochastic processes
            - Machine learning applications in trading""",
            model_name="claude-sonnet-4-20250514",
            dynamic_temperature_enabled=True,
            max_loops=request.max_loops,
            dynamic_context_window=True,
        )
        
        result = agent.run(task=request.task)
        
        return {
            "success": True,
            "result": str(result),
            "agent_name": "Quantitative-Trading-Agent",
            "task": request.task
        }
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Quantitative trading agent failed: {str(e)}")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

3. Run Your API

python agent_api.py

Or with uvicorn directly:

uvicorn agent_api:app --host 0.0.0.0 --port 8000 --reload

4. Test Your API

Your API will be available at:

• API: http://localhost:8000

• Documentation: http://localhost:8000/docs

• Alternative docs: http://localhost:8000/redoc

Usage Examples

Using curl

# Basic agent task
curl -X POST "http://localhost:8000/agent/run" \
     -H "Content-Type: application/json" \
     -d '{"task": "What are the best top 3 ETFs for gold coverage?"}'

# Quantitative trading agent
curl -X POST "http://localhost:8000/agent/quantitative-trading" \
     -H "Content-Type: application/json" \
     -d '{"task": "Analyze the current market conditions for gold ETFs"}'

Using Python requests

import requests

# Run basic agent
response = requests.post(
    "http://localhost:8000/agent/run",
    json={"task": "Explain quantum computing in simple terms"}
)
print(response.json())

# Run quantitative trading agent
response = requests.post(
    "http://localhost:8000/agent/quantitative-trading",
    json={"task": "What are the key factors affecting gold prices today?"}
)
print(response.json())

Advanced Configuration

Environment Variables

Create a .env file for configuration:

# .env
AGENT_MODEL_NAME=claude-sonnet-4-20250514
AGENT_MAX_LOOPS=3
API_HOST=0.0.0.0
API_PORT=8000
LOG_LEVEL=info

Enhanced Agent Factory

import os
from typing import Dict, Type
from swarms import Agent

class AgentFactory:
    """Factory for creating different types of agents"""
    
    AGENT_CONFIGS = {
        "default": {
            "agent_name": "Default-Agent",
            "agent_description": "Versatile AI agent for various tasks",
            "system_prompt": "You are a helpful AI assistant...",
            "model_name": "claude-sonnet-4-20250514"
        },
        "quantitative-trading": {
            "agent_name": "Quantitative-Trading-Agent",
            "agent_description": "Advanced quantitative trading agent",
            "system_prompt": "You are an expert quantitative trading agent...",
            "model_name": "claude-sonnet-4-20250514"
        },
        "research": {
            "agent_name": "Research-Agent",
            "agent_description": "Academic research and analysis agent",
            "system_prompt": "You are an expert research agent...",
            "model_name": "claude-sonnet-4-20250514"
        }
    }
    
    @classmethod
    def create_agent(cls, agent_type: str = "default", **kwargs) -> Agent:
        """Create an agent of the specified type"""
        if agent_type not in cls.AGENT_CONFIGS:
            raise ValueError(f"Unknown agent type: {agent_type}")
        
        config = cls.AGENT_CONFIGS[agent_type].copy()
        config.update(kwargs)
        
        return Agent(**config)

Authentication & Rate Limiting

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
import time

# Rate limiting
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

# Security
security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """Verify API token"""
    # Implement your token verification logic here
    if credentials.credentials != "your-secret-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token"
        )
    return credentials.credentials

@app.post("/agent/run", response_model=AgentResponse)
@limiter.limit("10/minute")
async def run_agent(
    request: AgentRequest,
    token: str = Depends(verify_token)
):
    """Run an agent with authentication and rate limiting"""
    # ... existing code ...

Production Deployment

Using Gunicorn

pip install gunicorn
gunicorn agent_api:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Using Docker

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "agent_api:app", "--host", "0.0.0.0", "--port", "8000"]

Using Docker Compose

version: '3.8'
services:
  agent-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - AGENT_MODEL_NAME=claude-sonnet-4-20250514
    volumes:
      - ./logs:/app/logs

Monitoring & Logging

Structured Logging

import logging
import json
from datetime import datetime

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request, call_next):
    """Log all requests and responses"""
    start_time = time.time()
    
    # Log request
    logger.info(f"Request: {request.method} {request.url}")
    
    response = await call_next(request)
    
    # Log response
    process_time = time.time() - start_time
    logger.info(f"Response: {response.status_code} - {process_time:.2f}s")
    
    return response

Health Checks

@app.get("/health/detailed")
async def detailed_health_check():
    """Detailed health check with agent status"""
    try:
        # Test agent creation
        agent = create_agent()
        
        return {
            "status": "healthy",
            "timestamp": datetime.utcnow().isoformat(),
            "agent_status": "available",
            "model_status": "connected"
        }
    except Exception as e:
        return {
            "status": "unhealthy",
            "timestamp": datetime.utcnow().isoformat(),
            "error": str(e)
        }

Best Practices

Best Practice	Description
Error Handling	Always wrap agent execution in try-catch blocks
Validation	Use Pydantic models for request validation
Rate Limiting	Implement rate limiting for production APIs
Authentication	Add proper authentication for sensitive endpoints
Logging	Log all requests and responses for debugging
Monitoring	Add health checks and metrics
Testing	Write tests for your API endpoints
Documentation	Keep your API documentation up to date

Troubleshooting

Common Issues

1. Port already in use: Change the port in the uvicorn command
2. Agent initialization fails: Check your API keys and model configuration
3. Memory issues: Reduce max_loops or implement streaming responses
4. Timeout errors: Increase timeout settings for long-running tasks

Performance Tips

Performance Tip	Description
Connection pooling	Reuse agent instances when possible
Async operations	Use async/await for I/O operations
Caching	Cache frequently requested responses
Load balancing	Use multiple worker processes for high traffic

Your FastAPI agent API is now ready to handle requests and scale with your needs!