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
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
@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
@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") ```