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