restapi

用途

REST API 技能提供全面的 REST API 设计与实现能力，面向 Golden Armada AI Agent Fleet Platform。

当处理以下内容时，可启用本技能：

API 端点设计
HTTP 请求/响应处理
API 认证
OpenAPI 文档
API 测试

主要能力

REST API 设计原则

资源命名应使用名词、复数形式：

Good - nouns, plural

GET    /agents
GET    /agents/{id}
POST   /agents
PUT    /agents/{id}
DELETE /agents/{id}

Nested resources

GET    /agents/{id}/tasks
POST   /agents/{id}/tasks

Bad - verbs, actions

GET    /getAgents
POST   /createAgent

HTTP 方法使用与幂等性：

Method	Usage	Idempotent
GET	Read resource	Yes
POST	Create resource	No
PUT	Replace resource	Yes
PATCH	Partial update	No
DELETE	Delete resource	Yes

状态码示例：

Success

200 OK              - Successful GET, PUT, PATCH
201 Created         - Successful POST
204 No Content      - Successful DELETE

Client Errors

400 Bad Request     - Validation error
401 Unauthorized    - Authentication required
403 Forbidden       - Not permitted
404 Not Found       - Resource not found
409 Conflict        - Duplicate/conflict
422 Unprocessable   - Semantic validation error

Server Errors

500 Internal Error  - Server error
502 Bad Gateway     - Upstream error
503 Service Unavailable - Temporary overload

API 实现（FastAPI）

from fastapi import FastAPI, HTTPException, Query, Path, Depends, status
from typing import List, Optional
from pydantic import BaseModel, Field

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

# Models

class AgentCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    type: str = Field(..., pattern="^(claude|gpt|gemini)$")

class AgentResponse(BaseModel):
    id: str
    name: str
    type: str
    status: str
    created_at: datetime

# Endpoints

@app.get("/agents", response_model=List[AgentResponse])
async def list_agents(
    skip: int = Query(0, ge=0),
    limit: int = Query(100, ge=1, le=1000),
    status: Optional[str] = Query(None)
):
    """List all agents with pagination and filtering."""
    return await agent_service.list(skip=skip, limit=limit, status=status)

@app.get("/agents/{agent_id}", response_model=AgentResponse)
async def get_agent(agent_id: str = Path(..., description="Agent ID")):
    """Get a specific agent by ID."""
    agent = await agent_service.get(agent_id)
    if not agent:
        raise HTTPException(status_code=404, detail="Agent not found")
    return agent

@app.post("/agents", response_model=AgentResponse, status_code=201)
async def create_agent(agent: AgentCreate):
    """Create a new agent."""
    return await agent_service.create(agent)

@app.put("/agents/{agent_id}", response_model=AgentResponse)
async def update_agent(agent_id: str, agent: AgentCreate):
    """Replace an existing agent."""
    existing = await agent_service.get(agent_id)
    if not existing:
        raise HTTPException(status_code=404, detail="Agent not found")
    return await agent_service.update(agent_id, agent)

@app.delete("/agents/{agent_id}", status_code=204)
async def delete_agent(agent_id: str):
    """Delete an agent."""
    deleted = await agent_service.delete(agent_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Agent not found")

认证

JWT 认证：

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    return await get_user(user_id)

# Protected endpoint

@app.get("/protected")
async def protected_route(user: User = Depends(get_current_user)):
    return {"user": user.email}

API Key 认证：

from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

async def verify_api_key(api_key: str = Depends(api_key_header)):
    if api_key != VALID_API_KEY:
        raise HTTPException(status_code=403, detail="Invalid API key")
    return api_key

错误处理

class ErrorResponse(BaseModel):
    error: str
    message: str
    details: Optional[dict] = None

@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
    return JSONResponse(
        status_code=400,
        content=ErrorResponse(
            error="validation_error",
            message=str(exc)
        ).dict()
    )

@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
    return JSONResponse(
        status_code=500,
        content=ErrorResponse(
            error="internal_error",
            message="An unexpected error occurred"
        ).dict()
    )

使用 curl 进行 API 测试

# GET

curl -X GET "http://localhost:8000/agents" \
  -H "Authorization: Bearer TOKEN"

# POST

curl -X POST "http://localhost:8000/agents" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer TOKEN" \
  -d '{"name": "agent-1", "type": "claude"}'

# PUT

curl -X PUT "http://localhost:8000/agents/123" \
  -H "Content-Type: application/json" \
  -d '{"name": "updated-agent", "type": "claude"}'

# DELETE

curl -X DELETE "http://localhost:8000/agents/123" \
  -H "Authorization: Bearer TOKEN"

OpenAPI 文档

from fastapi import FastAPI

app = FastAPI(
    title="Golden Armada API",
    description="AI Agent Fleet Platform API",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
    openapi_url="/openapi.json"
)

访问文档地址：

Swagger UI: http://localhost:8000/docs
ReDoc: http://localhost:8000/redoc
OpenAPI JSON: http://localhost:8000/openapi.json

适用场景

设计或评审 REST API 端点与资源命名
选择并正确使用 HTTP 方法与状态码
为 API 添加 JWT 或 API Key 认证
统一 API 错误响应格式
使用 curl 对接口进行手动测试
配置 OpenAPI/Swagger 文档

限制

关于该技能的收费、价格、平台支持范围、具体效果或额外使用步骤，资料中暂未提供。

技能介绍

用途

主要能力

REST API 设计原则

API 实现（FastAPI）

认证

错误处理

使用 curl 进行 API 测试

OpenAPI 文档

适用场景

限制