关于 fastapi
fastapi 是面向 FastAPI 应用开发者的技能，提供异步端点、Pydantic 模型、依赖注入、后台任务、WebSocket 和 API 文档相关模式与示例。
主要功能
- 01应用结构与运行命令
- 02异步端点和路由模式
- 03Pydantic 模型示例
- 04依赖注入模式
- 05后台任务与 WebSocket
适用场景
- 01创建 FastAPI 应用端点
- 02定义请求与响应模型
- 03通过依赖装配服务
- 04处理后台任务或 WebSocket 消息
安装
复制下面的命令，在终端中运行。
npx skills add https:/安装后在 Claude Code 中输入/fastapi确认已加载。
详细介绍
FastAPI 技能
为 Golden Armada AI Agent Fleet Platform 提供全面的 FastAPI 开发能力。
何时使用此技能
在以下场景中启用此技能：
- FastAPI 应用开发
- 异步端点实现
- Pydantic 模型定义
- 依赖注入模式
- OpenAPI/Swagger 文档
快速参考
运行命令
```bash
开发环境
uvicorn main:app --reload --host 0.0.0.0 --port 8000
生产环境
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4
使用 Gunicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000 ```
应用结构
``` 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 ```
基本应用
```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): # 启动 print("Starting up...") yield # 关闭 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 模型
```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)
```
路由器示例
```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) ): """创建新 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) ): """列出所有 Agent。""" 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) ): """根据 ID 获取 Agent。""" agent = await service.get(agent_id) if not agent: raise HTTPException(status_code=404, detail="Agent not found") return agent ```
依赖注入
```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) ```
后台任务
```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 支持
```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") ```