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

```bash

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

```python

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

```python

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

```python

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

```python

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

```python 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

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