FastAPI Agent API

pip install fastapi uvicorn swarms

from typing import Optional
import time

import uvicorn
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from swarms import Agent

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


class AgentRequest(BaseModel):
    task: str
    agent_name: Optional[str] = "default"
    max_loops: Optional[int] = 1
    temperature: Optional[float] = None


class AgentResponse(BaseModel):
    success: bool
    result: str
    agent_name: str
    task: str
    execution_time: Optional[float] = None


def create_agent(agent_name: str = "default") -> Agent:
    return Agent(
        agent_name=agent_name,
        agent_description="Versatile AI agent for various tasks",
        system_prompt=(
            "You are a helpful AI assistant. Be clear, accurate, and concise."
        ),
        model_name="claude-sonnet-4-20250514",
        dynamic_temperature_enabled=True,
        max_loops=1,
        dynamic_context_window=True,
    )


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


@app.get("/health")
async def 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):
    try:
        start_time = time.time()
        agent = create_agent(request.agent_name)
        result = agent.run(task=request.task, max_loops=request.max_loops)
        return AgentResponse(
            success=True,
            result=str(result),
            agent_name=request.agent_name,
            task=request.task,
            execution_time=time.time() - start_time,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent execution failed: {e}")


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

python agent_api.py

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

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?"}'

import requests

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

@app.post("/agent/quantitative-trading")
async def run_quant_agent(request: AgentRequest):
    try:
        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, statistical arbitrage, risk management, and "
                "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"Quant agent failed: {e}")

from swarms import Agent


class AgentFactory:
    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", **overrides) -> Agent:
        if agent_type not in cls.AGENT_CONFIGS:
            raise ValueError(f"Unknown agent type: {agent_type}")
        config = {**cls.AGENT_CONFIGS[agent_type], **overrides}
        return Agent(**config)

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

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

security = HTTPBearer()


def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    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_secure(
    request: AgentRequest,
    token: str = Depends(verify_token),
):
    ...

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

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"]

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