FastAPI Skill

Provides comprehensive FastAPI development capabilities for the Golden Armada AI Agent Fleet Platform.

When to Use This Skill

Activate this skill when working with:

• FastAPI application development

• Async endpoint implementation

• Pydantic model definitions

• Dependency injection patterns

• OpenAPI/Swagger documentation

Quick Reference

Run Commands

Development

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

Production

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

With Gunicorn

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000

Application Structure

app/
├── main.py
├── config.py
├── dependencies.py
├── routers/
│   ├── init.py
│   ├── agents.py
│   └── tasks.py
├── models/
│   ├── init.py
│   ├── agent.py
│   └── task.py
├── schemas/
│   ├── init.py
│   ├── agent.py
│   └── task.py
└── services/
├── init.py
└── llm_service.py

Basic Application

main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager

from app.routers import agents, tasks
from app.config import settings

@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
print("Starting up...")
yield
# Shutdown
print("Shutting down...")

app = FastAPI(
title="Golden Armada Agent API",
version="1.0.0",
lifespan=lifespan
)

app.add_middleware(
CORSMiddleware,
allow_origins=settings.allowed_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", "service": "golden-armada"}

Pydantic Schemas

schemas/agent.py

from pydantic import BaseModel, Field
from typing import Optional
from enum import Enum

class AgentType(str, Enum):
CLAUDE = "claude"
GPT = "gpt"
GEMINI = "gemini"

class AgentBase(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
type: AgentType
description: Optional[str] = None

class AgentCreate(AgentBase):
pass

class AgentResponse(AgentBase):
id: str
status: str
created_at: datetime

model_config = ConfigDict(from_attributes=True)

Router Example

routers/agents.py

from fastapi import APIRouter, HTTPException, Depends, status
from typing import List

from app.schemas.agent import AgentCreate, AgentResponse
from app.services.agent_service import AgentService
from app.dependencies import get_agent_service

router = APIRouter()

@router.post("/", response_model=AgentResponse, status_code=status.HTTP_201_CREATED)
async def create_agent(
agent: AgentCreate,
service: AgentService = Depends(get_agent_service)
):
"""Create a new agent."""
return await service.create(agent)

@router.get("/", response_model=List[AgentResponse])
async def list_agents(
skip: int = 0,
limit: int = 100,
service: AgentService = Depends(get_agent_service)
):
"""List all agents."""
return await service.list(skip=skip, limit=limit)

@router.get("/{agent_id}", response_model=AgentResponse)
async def get_agent(
agent_id: str,
service: AgentService = Depends(get_agent_service)
):
"""Get agent by ID."""
agent = await service.get(agent_id)
if not agent:
raise HTTPException(status_code=404, detail="Agent not found")
return agent

Dependency Injection

dependencies.py

from functools import lru_cache
from typing import Annotated
from fastapi import Depends

from app.config import Settings
from app.services.agent_service import AgentService
from app.services.llm_service import LLMService

@lru_cache
def get_settings():
return Settings()

async def get_llm_service(
settings: Annotated[Settings, Depends(get_settings)]
) -> LLMService:
return LLMService(api_key=settings.anthropic_api_key)

async def get_agent_service(
llm_service: Annotated[LLMService, Depends(get_llm_service)]
) -> AgentService:
return AgentService(llm_service=llm_service)

Background Tasks

from fastapi import BackgroundTasks

@router.post("/tasks/{task_id}/execute")
async def execute_task(
task_id: str,
background_tasks: BackgroundTasks,
service: TaskService = Depends(get_task_service)
):
background_tasks.add_task(service.execute_async, task_id)
return {"status": "accepted", "task_id": task_id}

WebSocket Support

from fastapi import WebSocket, WebSocketDisconnect

@router.websocket("/ws/{agent_id}")
async def agent_websocket(websocket: WebSocket, agent_id: str):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
response = await process_message(agent_id, data)
await websocket.send_text(response)
except WebSocketDisconnect:
print(f"Agent {agent_id} disconnected")