mcpField
R
restapi

restapi

github.com/thelobbi开发2157 次安装1 天前更新
代码仓库

关于 restapi

面向 Golden Armada AI Agent Fleet Platform 的 REST API 设计与实现 Skill。内容涵盖端点规范、FastAPI 示例、身份验证、错误处理、测试和 OpenAPI 文档。

主要功能

  1. 01资源命名与 HTTP 方法规范
  2. 02FastAPI 端点示例
  3. 03JWT 与 API 密钥验证
  4. 04错误响应处理
  5. 05使用 curl 测试 API
  6. 06配置 OpenAPI 文档

适用场景

  1. 01为 Agent 设计 REST 端点
  2. 02实现需要身份验证的 API 路由
  3. 03统一 API 错误与状态码
  4. 04使用 curl 测试请求

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/thelobbi/claude --skill restapi

安装后在 Claude Code 中输入/restapi确认已加载。

详细介绍

REST API 技能

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

何时使用此技能

在处理以下工作时启用此技能：

  • API 端点设计
  • HTTP 请求/响应处理
  • API 身份验证
  • OpenAPI 文档
  • API 测试

REST API 设计原则

资源命名

```

推荐 - 使用名词和复数形式

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

嵌套资源

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

不推荐 - 使用动词和动作

GET /getAgents POST /createAgent ```

HTTP 方法
方法用途幂等
GET读取资源是
POST创建资源否
PUT替换资源是
PATCH部分更新否
DELETE删除资源是
状态码

```

成功

200 OK - GET、PUT、PATCH 请求成功 201 Created - POST 请求成功 204 No Content - DELETE 请求成功

客户端错误

400 Bad Request - 验证错误 401 Unauthorized - 需要身份验证 403 Forbidden - 不允许执行 404 Not Found - 未找到资源 409 Conflict - 重复/冲突 422 Unprocessable - 语义验证错误

服务端错误

500 Internal Error - 服务端错误 502 Bad Gateway - 上游错误 503 Service Unavailable - 暂时过载 ```

API 实现（FastAPI）

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

模型

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

端点

@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) ): """列出所有 Agent，并支持分页和筛选。""" 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")): """根据 ID 获取指定 Agent。""" 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): """创建新的 Agent。""" return await agent_service.create(agent)

@app.put("/agents/{agent_id}", response_model=AgentResponse) async def update_agent(agent_id: str, agent: AgentCreate): """替换现有 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): """删除 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 密钥身份验证
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"
)

# Access documentation at:
# - Swagger UI: http://localhost:8000/docs
# - ReDoc: http://localhost:8000/redoc
# - OpenAPI JSON: http://localhost:8000/openapi.json

评论

支持 Markdown