API Patterns

# Collection
GET    /api/v1/documents        # List documents
POST   /api/v1/documents        # Create document

# Single resource
GET    /api/v1/documents/{id}   # Get document
PUT    /api/v1/documents/{id}   # Replace document
PATCH  /api/v1/documents/{id}   # Update document
DELETE /api/v1/documents/{id}   # Delete document

# Nested resources
GET    /api/v1/users/{id}/documents  # User's documents

{
  "data": {
    "id": "123",
    "type": "document",
    "attributes": {
      "title": "Example",
      "content": "..."
    }
  },
  "meta": {
    "total": 100,
    "page": 1,
    "per_page": 10
  }
}

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Invalid input",
    "details": [
      {"field": "title", "message": "Title is required"},
      {"field": "limit", "message": "Must be between 1 and 100"}
    ]
  }
}

from fastapi import FastAPI, HTTPException, Query, Path
from pydantic import BaseModel, Field

app = FastAPI(title="RAG-MCP API", version="1.0.0")

class InvalidSearchQuery(Exception):
    """Known request-level refusal from the application-owned search adapter."""

class SearchRequest(BaseModel):
    query: str = Field(..., min_length=1, description="Search query")
    limit: int = Field(10, ge=1, le=100, description="Max results")

class SearchResult(BaseModel):
    id: str
    title: str
    score: float
    content: str

class SearchResponse(BaseModel):
    results: list[SearchResult]
    total: int

@app.post("/api/v1/search", response_model=SearchResponse)
async def search(request: SearchRequest):
    """Search the knowledge base.

    Args:
        request: Search parameters

    Returns:
        Search results with scores
    """
    try:
        results = await perform_search(request.query, request.limit)
        return SearchResponse(results=results, total=len(results))
    except InvalidSearchQuery as exc:
        raise HTTPException(
            status_code=400,
            detail="The search query is not supported. Check its syntax.",
        ) from exc

class ListReposRequest(BaseModel):
    visibility: Literal["PUBLIC", "PRIVATE", "INTERNAL"] = Field(
        "PUBLIC",
        description=(
            "Repository visibility filter. "
            "PUBLIC = visible to anyone; "
            "PRIVATE = only members with explicit access; "
            "INTERNAL = visible to all org members (Enterprise only)."
        ),
    )

cursor: str | None = Field(
    None,
    description=(
        "Pagination cursor. Requires a `next_cursor` value obtained from a prior "
        "GET /api/v1/documents response. Omit on the first page; do not synthesize."
    ),
)

document_id: str = Field(
    ...,
    description=(
        "Exact document id, e.g. `doc_9f3a21`. Copy it verbatim from a search or "
        "list response — case-sensitive, do not type from memory or guess the format. "
        "Obtain it from GET /api/v1/documents or the search results."
    ),
)

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "search",
    "arguments": {"query": "test"}
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {"type": "text", "text": "Results..."}
    ]
  }
}

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32602,
    "message": "Invalid params",
    "data": {"field": "query", "message": "Required"}
  }
}

GET /api/v1/documents?page=2&per_page=20

{
  "data": [...],
  "meta": {
    "total": 150,
    "page": 2,
    "per_page": 20,
    "total_pages": 8
  }
}

GET /api/v1/documents?cursor=abc123&limit=20

{
  "data": [...],
  "meta": {
    "next_cursor": "xyz789",
    "has_more": true
  }
}

X-RateLimit-Limit: 100
X-RateLimit-Remaining: 95
X-RateLimit-Reset: 1699999999

from fastapi import Request
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.get("/api/v1/search")
@limiter.limit("100/minute")
async def search(request: Request, query: str):
    ...

from fastapi import Depends, HTTPException, Security
from fastapi.security import APIKeyHeader

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

async def verify_api_key(api_key: str = Security(api_key_header)):
    if api_key != settings.API_KEY:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return api_key

@app.get("/api/v1/protected")
async def protected_endpoint(api_key: str = Depends(verify_api_key)):
    ...

from fastapi import Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

async def verify_jwt(credentials: HTTPAuthorizationCredentials = Depends(security)):
    try:
        payload = jwt.decode(
            credentials.credentials,
            settings.JWT_SECRET,
            algorithms=["HS256"]
        )
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

/api/v1/documents
/api/v2/documents

Accept: application/vnd.myapi.v1+json