Too many requests
") # NEVER omit retryable field — forces agent to guess return {"type": "...", "status": 429, "detail": "Rate limited"} # NEVER use ambiguous categories return {"error_category": "error"} # Useless — be specific # NEVER put retry logic in prose only return {"what_you_should_do": "Please wait 30 seconds and try again"} # ↑ Missing retryable: true and retry_after: 30 — agent must parse prose ``` **Incorrect — status code only, agent must guess:** ```python return JSONResponse({"error": "Too many requests"}, status_code=429) ``` **Correct — structured fields, agent branches deterministically:** ```python return JSONResponse( content={ "type": "https://api.example.com/problems/rate-limit-exceeded", "title": "Rate Limit Exceeded", "status": 429, "error_category": "rate_limit", "retryable": True, "retry_after": 30, "owner_action_required": False, "what_you_should_do": "Wait 30 seconds, then retry.", }, status_code=429, media_type="application/problem+json", ) ``` ### Key Rules - Always include `error_category` and `retryable` on every error response - Set `retry_after` (seconds) whenever `retryable` is true - Set `owner_action_required` when the API consumer cannot self-resolve - Keep `what_you_should_do` under 50 words — it enters the LLM context - Use `Accept` header content negotiation: `application/problem+json` for agents, `text/markdown` for LLMs, `text/html` for browsers - Design 8-15 error categories per domain — enough for deterministic branching, few enough to be learnable - Agents should send `Accept: application/problem+json, */*` to signal structured error preference ### Keep OpenAPI specifications complete and up-to-date as the API provider-consumer contract — HIGH ## OpenAPI Specifications Patterns for creating comprehensive OpenAPI 3.1 specifications with proper schema definitions, authentication, and error documentation. **OpenAPI 3.1 Structure:** ```yaml openapi: 3.1.0 info: title: Your API Name version: 1.0.0 description: | Brief description of what this API does. ## Authentication This API uses Bearer tokens for authentication. ## Rate Limiting - 1000 requests per hour per API key servers: - url: https://api.company.com/v1 description: Production server - url: http://localhost:3000/v1 description: Local development ``` **Endpoint Documentation with FastAPI:** ```python @router.get( "/analyses/{analysis_id}", responses={ 404: {"model": ErrorResponse, "description": "Analysis not found"}, 500: {"model": ErrorResponse, "description": "Internal server error"}, }, summary="Get analysis details", description="Retrieve detailed information about a specific analysis", ) async def get_analysis( analysis_id: Annotated[uuid.UUID, Path(description="Analysis UUID")] ) -> AnalysisResponse: ... ``` **Reusable Components:** ```yaml components: securitySchemes: bearerAuth: type: http scheme: bearer bearerFormat: JWT parameters: PageParam: name: page in: query schema: type: integer minimum: 1 default: 1 PerPageParam: name: per_page in: query schema: type: integer minimum: 1 maximum: 100 default: 20 headers: X-RateLimit-Limit: description: Maximum requests allowed per hour schema: type: integer example: 1000 X-RateLimit-Remaining: description: Requests remaining in current window schema: type: integer schemas: Error: type: object required: [error] properties: error: type: object required: [code, message] properties: code: type: string example: "VALIDATION_ERROR" message: type: string details: type: array items: type: object properties: field: type: string message: type: string request_id: type: string responses: NotFoundError: description: Resource not found content: application/json: schema: $ref: '#/components/schemas/Error' ValidationError: description: Validation failed content: application/json: schema: $ref: '#/components/schemas/Error' ``` **Per-Version OpenAPI Docs:** ```python from fastapi import FastAPI from fastapi.openapi.utils import get_openapi def custom_openapi_v1(): return get_openapi( title="OrchestKit API", version="1.0.0", routes=v1_router.routes, ) def custom_openapi_v2(): return get_openapi( title="OrchestKit API", version="2.0.0", routes=v2_router.routes, ) app.mount("/docs/v1", create_docs_app(custom_openapi_v1)) app.mount("/docs/v2", create_docs_app(custom_openapi_v2)) ``` **Pydantic Schema Validation:** ```python from pydantic import BaseModel, HttpUrl, Field class AnalyzeRequest(BaseModel): url: HttpUrl analysis_id: str | None = None skill_level: str = Field( default="beginner", pattern="^(beginner|intermediate|advanced)$", ) ``` **Incorrect — Missing response documentation:** ```python # No response schema or error docs @router.get("/analyses/{id}") async def get_analysis(id: str): return await service.get(id) ``` **Correct — Full OpenAPI documentation:** ```python @router.get( "/analyses/{id}", responses={ 404: {"model": ErrorResponse, "description": "Analysis not found"}, 500: {"model": ErrorResponse, "description": "Internal error"} }, summary="Get analysis details" ) async def get_analysis(id: Annotated[str, Path(description="Analysis UUID")]) -> AnalysisResponse: return await service.get(id) ``` **Key rules:** - Use OpenAPI 3.1 for all new API specifications - Define reusable schemas, parameters, and responses in `components` - Document all error responses with examples - Include security schemes and rate limit headers - Generate per-version documentation when versioning ### Model REST resources with proper nesting and filters to minimize client round-trips — HIGH ## Resource Modeling Patterns for modeling API resources with hierarchical relationships, filtering, sorting, and field selection. **Hierarchical Relationships:** ``` # Express ownership and containment through URL hierarchy GET /api/v1/analyses/{analysis_id}/artifact GET /api/v1/teams/{team_id}/members POST /api/v1/projects/{project_id}/tasks # NOT query params for primary relationships GET /api/v1/artifact?analysis_id={id} # Avoid GET /api/v1/analysis_artifact/{id} # Avoid ``` **Query Parameter Filtering:** ```python @router.get("/analyses") async def list_analyses( status: str | None = None, content_type: str | None = None, created_after: datetime | None = None, created_before: datetime | None = None, ) -> list[AnalysisResponse]: filters = {} if status: filters["status"] = status if content_type: filters["content_type"] = content_type return await repo.find_all(filters=filters) ``` **Usage:** ``` GET /api/v1/analyses?status=completed&content_type=article GET /api/v1/analyses?created_after=2025-01-01&created_before=2025-12-31 ``` **Sorting:** ```python @router.get("/analyses") async def list_analyses( sort: str = Query(default="-created_at"), ) -> list[AnalysisResponse]: direction = "desc" if sort.startswith("-") else "asc" field = sort.lstrip("-") return await repo.find_all(order_by=field, direction=direction) ``` **Usage:** ``` GET /api/v1/analyses?sort=-created_at # Newest first GET /api/v1/analyses?sort=title # Alphabetical ``` **Field Selection (Sparse Fieldsets):** ```python @router.get("/analyses") async def list_analyses( fields: str | None = None, ) -> list[dict[str, Any]]: selected_fields = fields.split(",") if fields else None results = await repo.find_all() if selected_fields: return [ {k: v for k, v in item.model_dump().items() if k in selected_fields} for item in results ] return results ``` **Usage:** ``` GET /api/v1/analyses?fields=id,title,status ``` **GraphQL Connection Pattern (for GraphQL APIs):** ```graphql type Query { users(first: Int, after: String): UserConnection! } type UserConnection { edges: [UserEdge!]! pageInfo: PageInfo! totalCount: Int! } type UserEdge { node: User! cursor: String! } type PageInfo { hasNextPage: Boolean! hasPreviousPage: Boolean! startCursor: String endCursor: String } ``` **Best Practices:** ```python # Empty collections: Return empty array, not null {"data": []} # Correct {"data": null} # Wrong # Deleted resources: Return 404, not null # 404 Not Found (correct) # {"data": null} (wrong) # Null fields: Be explicit {"title": null, "description": ""} # Clear intent ``` **Incorrect — Flat URLs with query params for hierarchy:** ```typescript // Parent relationship in query param GET /api/v1/artifacts?analysis_id=abc-123 ``` **Correct — Hierarchical URLs:** ```typescript // Express ownership in URL structure GET /api/v1/analyses/abc-123/artifacts ``` **Key rules:** - Use hierarchical URLs for parent-child relationships - Support filtering via query parameters on list endpoints - Use `-field` prefix for descending sort order - Return empty arrays (not null) for empty collections - Provide field selection for bandwidth optimization ### Follow REST conventions for naming, HTTP methods, and status codes consistently — HIGH ## REST API Conventions Standard conventions for RESTful API design covering resource naming, HTTP methods, status codes, and pagination. **Resource Naming:** ``` # Plural nouns for collections GET /users GET /users/123 POST /users # Hierarchical relationships GET /users/123/orders # Orders for specific user GET /teams/5/members # Members of specific team POST /projects/10/tasks # Create task in project 10 # Kebab-case for multi-word resources /shopping-carts /order-items /user-preferences ``` **HTTP Methods:** | Method | Purpose | Idempotent | Safe | Example | |--------|---------|------------|------|---------| | GET | Retrieve resource(s) | Yes | Yes | `GET /users/123` | | POST | Create resource | No | No | `POST /users` | | PUT | Replace entire resource | Yes | No | `PUT /users/123` | | PATCH | Partial update | No | No | `PATCH /users/123` | | DELETE | Remove resource | Yes | No | `DELETE /users/123` | **Status Codes:** | Code | Name | Use Case | |------|------|----------| | 200 | OK | Successful GET, PUT, PATCH | | 201 | Created | Successful POST (include `Location` header) | | 204 | No Content | Successful DELETE | | 400 | Bad Request | Invalid request syntax | | 401 | Unauthorized | Missing or invalid auth | | 403 | Forbidden | Authenticated but not authorized | | 404 | Not Found | Resource doesn't exist | | 409 | Conflict | Duplicate/constraint violation | | 422 | Unprocessable | Validation failed | | 429 | Too Many Requests | Rate limit exceeded | | 500 | Internal Error | Server error | **Cursor-Based Pagination (Recommended):** ```python @router.get("/analyses") async def list_analyses( cursor: str | None = None, limit: int = Query(default=20, le=100) ) -> PaginatedResponse: results = await repo.get_paginated(cursor=cursor, limit=limit) return { "data": results, "pagination": { "next_cursor": encode_cursor(results[-1].id) if results else None, "has_more": len(results) == limit } } ``` **Common Pitfalls:** | Pitfall | Bad | Good | |---------|-----|------| | Verbs in URLs | `POST /createUser` | `POST /users` | | Inconsistent naming | `/users, /userOrders` | `/users, /orders` | | Ignoring HTTP methods | `POST /users/123/delete` | `DELETE /users/123` | | Exposing internals | `/users-table` | `/users` | | Generic errors | `"Something went wrong"` | `"Email already exists"` | **Incorrect — Verbs in URLs:** ```python # RPC-style endpoints POST /createUser POST /users/123/delete GET /getUserOrders?id=123 ``` **Correct — REST conventions:** ```python # Resource-oriented POST /users DELETE /users/123 GET /users/123/orders ``` **Key rules:** - Always use plural nouns for resources - Use kebab-case for multi-word resource names - Map CRUD to proper HTTP methods - Include `Location` header in 201 responses - Use cursor-based pagination for large datasets ### GraphQL Schema Patterns and FastAPI Integration — HIGH ## GraphQL Schema Patterns and FastAPI Integration **Incorrect -- unprotected GraphQL endpoints:** ```python # No authentication or authorization @strawberry.type class Query: @strawberry.field async def all_users(self, info: strawberry.Info) -> list[User]: return await info.context.user_service.list_all() # Anyone can see all users! # Exposing internal IDs @strawberry.type class User: id: int # Exposes auto-increment ID! ``` **Correct -- permission classes for authorization:** ```python from strawberry.permission import BasePermission class IsAuthenticated(BasePermission): message = "User is not authenticated" async def has_permission(self, source, info: strawberry.Info, **kwargs) -> bool: return info.context.current_user_id is not None class IsAdmin(BasePermission): message = "Admin access required" async def has_permission(self, source, info: strawberry.Info, **kwargs) -> bool: user_id = info.context.current_user_id if not user_id: return False user = await info.context.user_service.get(user_id) return user and user.role == "admin" @strawberry.type class Query: @strawberry.field(permission_classes=[IsAuthenticated]) async def me(self, info: strawberry.Info) -> User: return await info.context.user_service.get(info.context.current_user_id) @strawberry.field(permission_classes=[IsAdmin]) async def all_users(self, info: strawberry.Info) -> list[User]: return await info.context.user_service.list_all() ``` **Correct -- FastAPI integration with context getter:** ```python from strawberry.fastapi import GraphQLRouter schema = strawberry.Schema(query=Query, mutation=Mutation, subscription=Subscription) async def get_context(request: Request, user_service=Depends(get_user_service)) -> GraphQLContext: return GraphQLContext(request=request, user_service=user_service) graphql_router = GraphQLRouter(schema, context_getter=get_context, graphiql=True) app = FastAPI() app.include_router(graphql_router, prefix="/graphql") ``` **Correct -- subscriptions with Redis PubSub:** ```python @strawberry.type class Subscription: @strawberry.subscription async def user_updated(self, info: strawberry.Info, user_id: strawberry.ID) -> AsyncGenerator[User, None]: async for message in info.context.pubsub.subscribe(f"user:{user_id}:updated"): yield User(**message) ``` Key decisions: - Use opaque IDs (strawberry.ID) not internal auto-increment - Permission classes for field-level authorization - Redis PubSub for subscription horizontal scaling - Context getter for dependency injection ### Design type-safe GraphQL schemas with Strawberry to prevent N+1 query problems — HIGH ## Strawberry GraphQL Schema Design **Incorrect -- N+1 queries in resolvers:** ```python # Making database calls in resolver loops @strawberry.type class Post: author_id: strawberry.ID @strawberry.field async def author(self, info: strawberry.Info) -> "User": # N+1: One query per post! return await db.get_user(self.author_id) ``` **Correct -- DataLoader for batched loading:** ```python from strawberry.dataloader import DataLoader class UserLoader(DataLoader[str, "User"]): def __init__(self, user_repo): super().__init__(load_fn=self.batch_load) self.user_repo = user_repo async def batch_load(self, keys: list[str]) -> list["User"]: users = await self.user_repo.get_many(keys) user_map = {u.id: u for u in users} return [user_map.get(key) for key in keys] @strawberry.type class Post: author_id: strawberry.ID @strawberry.field async def author(self, info: strawberry.Info) -> "User": return await info.context.user_loader.load(self.author_id) ``` **Correct -- type-safe schema with Private fields:** ```python import strawberry from strawberry import Private @strawberry.type class User: id: strawberry.ID email: str name: str password_hash: Private[str] # Not exposed in schema @strawberry.field def display_name(self) -> str: return f"{self.name} ({self.email})" @strawberry.input class CreateUserInput: email: str name: str password: str ``` **Correct -- union types for mutation error handling:** ```python @strawberry.type class CreateUserSuccess: user: User @strawberry.type class UserError: message: str code: str field: str | None = None @strawberry.type class CreateUserError: errors: list[UserError] CreateUserResult = strawberry.union("CreateUserResult", [CreateUserSuccess, CreateUserError]) ``` Key decisions: - Schema approach: Code-first with Strawberry types - N+1 prevention: DataLoader for ALL nested resolvers - Pagination: Relay-style cursor pagination - Auth: Permission classes (IsAuthenticated, IsAdmin) - Errors: Union types for mutations - Use when: Complex data relationships, client-driven fetching, real-time subscriptions - Do NOT use when: Simple CRUD (use REST), internal microservices (use gRPC) ### Define and implement gRPC services with compile-time type safety for microservices — HIGH ## gRPC Service Definition and Implementation **Incorrect -- REST for internal service communication:** ```python # Using REST for high-frequency internal calls response = requests.post("http://user-service/api/users", json={"email": email, "name": name}) # High serialization overhead, no compile-time validation, manual error mapping ``` **Correct -- protobuf service definition:** ```text syntax = "proto3"; package user.v1; import "google/protobuf/timestamp.proto"; service UserService { rpc GetUser(GetUserRequest) returns (User); rpc CreateUser(CreateUserRequest) returns (User); rpc ListUsers(ListUsersRequest) returns (stream User); // Server streaming rpc BulkCreateUsers(stream CreateUserRequest) returns (BulkCreateResponse); // Client streaming } message User { string id = 1; string email = 2; string name = 3; UserStatus status = 4; google.protobuf.Timestamp created_at = 5; } enum UserStatus { USER_STATUS_UNSPECIFIED = 0; USER_STATUS_ACTIVE = 1; USER_STATUS_INACTIVE = 2; } ``` **Correct -- async server implementation:** ```python import grpc.aio from app.protos import user_service_pb2 as pb2, user_service_pb2_grpc as pb2_grpc class UserServiceServicer(pb2_grpc.UserServiceServicer): async def GetUser(self, request, context): user = await self.user_repo.get(request.user_id) if not user: await context.abort(grpc.StatusCode.NOT_FOUND, f"User {request.user_id} not found") return self._to_proto(user) async def CreateUser(self, request, context): if not request.email or "@" not in request.email: await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Invalid email") user = await self.user_repo.create(email=request.email, name=request.name) return self._to_proto(user) ``` **Correct -- client with timeout and retry:** ```python class UserServiceClient: def __init__(self, host: str = "localhost:50051"): self.channel = grpc.insecure_channel(host, options=[ ("grpc.keepalive_time_ms", 30000), ("grpc.keepalive_timeout_ms", 10000), ]) self.stub = pb2_grpc.UserServiceStub(self.channel) def get_user(self, user_id: str, timeout: float = 5.0): try: return self.stub.GetUser(pb2.GetUserRequest(user_id=user_id), timeout=timeout) except grpc.RpcError as e: if e.code() == grpc.StatusCode.NOT_FOUND: raise UserNotFoundError(user_id) raise ``` Key decisions: - Proto organization: One service per file, shared messages in common.proto - Versioning: Package version (user.v1, user.v2), backward compatible - Always set client-side deadlines (timeout) - Always include health checks for load balancers - Use when: Internal microservices, streaming, polyglot, strong typing needed - Do NOT use when: Public APIs (use REST/GraphQL), simple CRUD, no HTTP/2 ### Implement gRPC streaming patterns and interceptors for real-time data and observability — HIGH ## gRPC Streaming and Interceptors **Incorrect -- ignoring stream cancellation:** ```python # Client may disconnect but server keeps processing def ListUsers(self, request, context): for user in all_users: yield self._to_proto(user) # No cancellation check! ``` **Correct -- server streaming with cancellation check:** ```python def ListUsers(self, request, context): """Server streaming: yield users one by one.""" for user in self.user_repo.iterate(page_size=request.page_size or 100): if not context.is_active(): # Check if client disconnected return yield self._to_proto(user) ``` **Correct -- bidirectional streaming:** ```python async def UserUpdates(self, request_iterator, context): """Bidirectional: receive updates, yield results.""" async for request in request_iterator: if not context.is_active(): return user = await self.user_repo.update(request.user_id, request.changes) yield self._to_proto(user) ``` **Correct -- auth interceptor:** ```python class AuthInterceptor(grpc.ServerInterceptor): def __init__(self, auth_service): self.auth_service = auth_service self.public_methods = {"/user.v1.UserService/CreateUser"} def intercept_service(self, continuation, handler_call_details): if handler_call_details.method not in self.public_methods: metadata = dict(handler_call_details.invocation_metadata) token = metadata.get("authorization", "").replace("Bearer ", "") if not token or not self.auth_service.verify(token): return grpc.unary_unary_rpc_method_handler( lambda req, ctx: ctx.abort(grpc.StatusCode.UNAUTHENTICATED, "Invalid token") ) return continuation(handler_call_details) ``` **Correct -- client retry interceptor with exponential backoff:** ```python class RetryInterceptor(grpc.UnaryUnaryClientInterceptor): def __init__(self, max_retries: int = 3): self.max_retries = max_retries self.retry_codes = {grpc.StatusCode.UNAVAILABLE, grpc.StatusCode.DEADLINE_EXCEEDED} def intercept_unary_unary(self, continuation, client_call_details, request): for attempt in range(self.max_retries): try: return continuation(client_call_details, request) except grpc.RpcError as e: if e.code() not in self.retry_codes or attempt == self.max_retries - 1: raise time.sleep(2 ** attempt) # Exponential backoff ``` Key decisions: - Server streaming: Always check `context.is_active()` before yielding - Auth: Interceptor with metadata, JWT tokens, public method allowlist - Retry: Exponential backoff, only retry UNAVAILABLE and DEADLINE_EXCEEDED - Always close channels to prevent resource leaks - Never skip deadline/timeout on client calls ### Integrate messaging platforms securely with webhook validation and delivery guarantees — HIGH # Messaging Platform Integrations ## Platform Selection | Platform | API Style | Best For | Limitations | |----------|-----------|----------|-------------| | WhatsApp (WAHA) | REST + Webhooks | Business messaging, notifications | Session management, rate limits | | Telegram Bot API | REST + Webhooks/Polling | Interactive bots, commands | 30 msg/sec per bot | | Slack | REST + Events API | Team workflows, notifications | Workspace-scoped | ## WhatsApp via WAHA ```typescript // Send message via WAHA (WhatsApp HTTP API) const response = await fetch(`${WAHA_URL}/api/sendText`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ chatId: `${phone}@c.us`, text: message, session: "default", }), }); ``` ## Telegram Bot API ```typescript // Set webhook for Telegram bot await fetch(`https://api.telegram.org/bot${TOKEN}/setWebhook`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ url: `${BASE_URL}/api/telegram/webhook`, secret_token: WEBHOOK_SECRET, }), }); ``` ## Webhook Security (Critical) Every platform provides a signature verification mechanism. **Always verify.** ```typescript // Telegram: verify secret_token header function verifyTelegramWebhook(req: Request): boolean { return req.headers["x-telegram-bot-api-secret-token"] === WEBHOOK_SECRET; } ``` - **WhatsApp WAHA**: API key header authentication - **Telegram**: `secret_token` in webhook registration, verified via header - **Slack**: HMAC-SHA256 signing secret verification ## Anti-Patterns **Incorrect:** - Processing webhooks without signature verification — anyone can POST fake messages - Synchronous processing of incoming messages — webhook timeout causes retries - No idempotency on webhook handlers — duplicate messages on retry **Correct:** - Always verify webhook signatures before processing - Acknowledge webhook immediately (200 OK), process async via queue - Store message IDs and deduplicate on retry - Rate-limit outgoing messages per platform limits ## References - `references/whatsapp-waha.md` — WAHA setup, session lifecycle, message types - `references/telegram-bot-api.md` — Bot setup, webhook config, keyboard patterns - `references/webhook-security.md` — Signature verification patterns per platform ### Configure Payload CMS 3.0 collections and access control patterns for Next.js — HIGH # Payload CMS 3.0 Patterns ## CMS Selection Decision Tree | Factor | Payload | Sanity | Strapi | |--------|---------|--------|--------| | Runtime | Next.js (self-hosted) | Hosted (GROQ API) | Node.js (self-hosted) | | TypeScript | First-class, generated types | Plugin-based | Partial | | Data ownership | Full (your DB) | Sanity cloud | Full (your DB) | | Admin UI | Customizable React | Sanity Studio | Built-in | | Best for | Next.js apps, developer-owned | Content teams, editorial | REST-first APIs | **Choose Payload when**: Next.js project, need full data ownership, developer-first workflow. **Choose Sanity when**: Content-heavy editorial teams, need hosted GROQ API, real-time collaboration. ## Collection Design ```typescript // Payload 3.0 collection config import type { CollectionConfig } from "payload"; export const Posts: CollectionConfig = { slug: "posts", admin: { useAsTitle: "title" }, access: { read: () => true, create: ({ req: { user } }) => Boolean(user), update: ({ req: { user } }) => user?.role === "admin", delete: ({ req: { user } }) => user?.role === "admin", }, fields: [ { name: "title", type: "text", required: true }, { name: "content", type: "richText" }, { name: "author", type: "relationship", relationTo: "users" }, { name: "status", type: "select", options: ["draft", "published"] }, ], hooks: { beforeChange: [({ data }) => ({ ...data, updatedAt: new Date() })], }, }; ``` ## Access Control Patterns - **Collection-level**: `read`, `create`, `update`, `delete` functions - **Field-level**: Per-field `access` for sensitive data - **Role-based**: Check `user.role` in access functions - **Local API**: Uses `overrideAccess: true` by default — be explicit when calling from server ## Anti-Patterns **Incorrect:** - Using Local API without `overrideAccess: false` in user-facing code — bypasses all access control - Putting business logic in hooks instead of service layer — untestable - Storing large files in the database — use S3/R2 upload adapter **Correct:** - Always set `overrideAccess: false` in API routes that serve user requests - Keep hooks thin — validate/transform only, delegate to services - Configure upload collections with S3-compatible storage adapter ## References - `references/payload-vs-sanity.md` — Detailed comparison, decision matrix, migration paths - Collection design (field types, relationships, blocks, validation) and access control (RBAC, field-level, multi-tenant) are first-party documented: https://payloadcms.com/docs ### Stream server-sent events with auto-reconnect for LLM responses and notifications — HIGH ## Server-Sent Events (SSE) Streaming **Incorrect -- no keepalive or cleanup:** ```python # No keepalive, no abort handling, no reconnection support @app.get("/stream") async def stream(): async def generate(): for item in data: yield f"data: {item}\n\n" return StreamingResponse(generate(), media_type="text/event-stream") ``` **Correct -- SSE with keepalive and abort handling (Next.js):** ```typescript export async function GET(req: Request) { const encoder = new TextEncoder() const stream = new ReadableStream({ async start(controller) { controller.enqueue(encoder.encode('data: Hello\n\n')) // Keep connection alive every 30s const interval = setInterval(() => { controller.enqueue(encoder.encode(': keepalive\n\n')) }, 30000) // Cleanup on client disconnect req.signal.addEventListener('abort', () => { clearInterval(interval) controller.close() }) } }) return new Response(stream, { headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', } }) } ``` **Correct -- LLM token streaming pattern:** ```typescript export async function POST(req: Request) { const { messages } = await req.json() const stream = await openai.chat.completions.create({ model: 'gpt-5.5', messages, stream: true }) const encoder = new TextEncoder() return new Response( new ReadableStream({ async start(controller) { for await (const chunk of stream) { const content = chunk.choices[0]?.delta?.content if (content) { controller.enqueue(encoder.encode("data: " + JSON.stringify({ content }) + "\n\n")) } } controller.enqueue(encoder.encode('data: [DONE]\n\n')) controller.close() } }), { headers: { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache' } } ) } ``` **Correct -- reconnecting client with exponential backoff:** ```typescript class ReconnectingEventSource { private eventSource: EventSource | null = null private reconnectDelay = 1000 private maxReconnectDelay = 30000 constructor(private url: string, private onMessage: (data: string) => void) { this.connect() } private connect() { this.eventSource = new EventSource(this.url) this.eventSource.onmessage = (event) => { this.reconnectDelay = 1000 this.onMessage(event.data) } this.eventSource.onerror = () => { this.eventSource?.close() setTimeout(() => this.connect(), this.reconnectDelay) this.reconnectDelay = Math.min(this.reconnectDelay * 2, this.maxReconnectDelay) } } } ``` Key decisions: - SSE for one-way server-to-client (LLM streaming, notifications) - Keepalive every 30s to prevent timeouts - Handle browser 6-connection-per-domain limit (use HTTP/2) - Exponential backoff for reconnection (1s to 30s) ### Implement WebSocket bidirectional streaming with async generator cleanup for resource safety — HIGH ## WebSocket and Async Generator Patterns **Incorrect -- no message validation or heartbeat:** ```python # No heartbeat, no validation, no reconnection wss.on('connection', (ws) => { ws.on('message', (data) => { wss.clients.forEach((client) => client.send(data)) // Raw broadcast! }) }) ``` **Correct -- WebSocket with heartbeat and validation:** ```typescript const wss = new WebSocketServer({ port: 8080 }) wss.on('connection', (ws) => { // Heartbeat const heartbeat = setInterval(() => ws.ping(), 30000) ws.on('message', (data) => { const parsed = JSON.parse(data.toString()) // Validate message structure if (!parsed.type || !parsed.text) return // Broadcast to connected clients wss.clients.forEach((client) => { if (client !== ws && client.readyState === WebSocket.OPEN) { client.send(JSON.stringify(parsed)) } }) }) ws.on('close', () => clearInterval(heartbeat)) }) ``` **Incorrect -- async generator without cleanup:** ```python # Generator not closed if exception occurs mid-iteration async for chunk in external_api_stream(): # Resource leak if exception! yield process(chunk) ``` **Correct -- aclosing() for guaranteed async generator cleanup:** ```python from contextlib import aclosing # Guaranteed cleanup with aclosing() async def stream_llm_response(prompt: str): async with aclosing(llm.astream(prompt)) as stream: async for chunk in stream: yield chunk.content # Consumption with proper cleanup async def consume(): async with aclosing(stream_llm_response("Hello")) as response: async for token in response: handle(token) ``` **When to use aclosing():** | Scenario | Use aclosing() | |----------|----------------| | External API streaming (LLM, HTTP) | Always | | Database streaming results | Always | | File streaming | Always | | Simple in-memory generators | Optional | | Generator with try/finally cleanup | Always | Key decisions: - WebSocket for bidirectional real-time (chat, collaboration) - SSE for one-way server-to-client (use SSE rule instead) - Always implement heartbeat/ping-pong for WebSockets - Always use `aclosing()` for external resource async generators - Implement backpressure with ReadableStream flow control - Monitor buffer sizes, pause production when consumer is slow ### Implement header-based API versioning with clean URLs and content negotiation — HIGH ## Header-Based Versioning Version selection via HTTP headers for clean URLs, best suited for internal APIs. **X-API-Version Header:** ```python from fastapi import Header, HTTPException, Depends SUPPORTED_VERSIONS = {1, 2} DEFAULT_VERSION = 2 async def get_api_version( x_api_version: str = Header(default="1", alias="X-API-Version") ) -> int: try: version = int(x_api_version) if version not in SUPPORTED_VERSIONS: raise ValueError() return version except ValueError: raise HTTPException( 400, f"Invalid API version. Supported: {SUPPORTED_VERSIONS}", ) @router.get("/users/{user_id}") async def get_user( user_id: str, version: int = Depends(get_api_version), service: UserService = Depends(), ): user = await service.get_user(user_id) if version == 1: return UserResponseV1(id=user.id, name=user.full_name) else: return UserResponseV2( id=user.id, first_name=user.first_name, last_name=user.last_name, ) ``` **Content Negotiation (Media Type Versioning):** ```python from fastapi import Request MEDIA_TYPES = { "application/vnd.orchestkit.v1+json": 1, "application/vnd.orchestkit.v2+json": 2, "application/json": 2, # Default to latest } async def get_version_from_accept(request: Request) -> int: accept = request.headers.get("Accept", "application/json") return MEDIA_TYPES.get(accept, 2) @router.get("/users/{user_id}") async def get_user( user_id: str, version: int = Depends(get_version_from_accept), ): ... ``` **Testing Header Versioning:** ```python import pytest from httpx import AsyncClient @pytest.mark.asyncio async def test_header_versioning(client: AsyncClient): # Request with v1 header response = await client.get( "/api/users/123", headers={"X-API-Version": "1"}, ) data = response.json() assert "avatar_url" not in data # Request with v2 header response = await client.get( "/api/users/123", headers={"X-API-Version": "2"}, ) data = response.json() assert "avatar_url" in data ``` **When to Use Header vs URL Path:** | Criteria | URL Path | Header | |----------|----------|--------| | **Visibility** | Clear in URL | Hidden in headers | | **Testing** | Easy with browser/curl | Needs header tools | | **Caching** | CDN-friendly | Requires Vary header | | **Best for** | Public APIs | Internal APIs | | **Multiple versions** | Separate route trees | Single route tree | **Incorrect — No default version:** ```python # Breaks when header missing async def get_version(x_api_version: str = Header()): return int(x_api_version) # Error if header absent! ``` **Correct — Default to latest version:** ```python # Falls back to latest stable version async def get_version( x_api_version: str = Header(default="2") ) -> int: return int(x_api_version) ``` **Key rules:** - Default to latest stable version when header is absent - Validate version against a supported versions set - Return 400 with helpful message for unsupported versions - Use header versioning only for internal APIs - Always document which versions are supported ### Implement URL path versioning for public APIs without routing conflicts or code duplication — HIGH ## URL Path Versioning The recommended versioning strategy for public APIs using URL path prefixes. **FastAPI Directory Structure:** ``` backend/app/ ├── api/ │ ├── v1/ │ │ ├── __init__.py │ │ ├── routes/ │ │ │ ├── users.py │ │ │ └── analyses.py │ │ └── schemas/ │ │ ├── user.py │ │ └── analysis.py │ ├── v2/ │ │ ├── __init__.py │ │ ├── routes/ │ │ │ ├── users.py # Updated schemas │ │ │ └── analyses.py │ │ └── schemas/ │ │ ├── user.py # New schema version │ │ └── analysis.py │ └── router.py # Combines all versions ├── core/ └── services/ # Shared across versions ``` **Router Setup:** ```python # backend/app/api/router.py from fastapi import APIRouter from app.api.v1.router import router as v1_router from app.api.v2.router import router as v2_router api_router = APIRouter() api_router.include_router(v1_router, prefix="/v1") api_router.include_router(v2_router, prefix="/v2") # main.py app.include_router(api_router, prefix="/api") ``` **Version-Specific Schemas:** ```python # v1/schemas/user.py class UserResponseV1(BaseModel): id: str name: str # Single name field # v2/schemas/user.py class UserResponseV2(BaseModel): id: str first_name: str # Split into first/last last_name: str full_name: str # Computed for convenience ``` **Shared Business Logic (Version-Agnostic Services):** ```python # services/user_service.py (version-agnostic) class UserService: async def get_user(self, user_id: str) -> User: return await self.repo.get_by_id(user_id) # v1/routes/users.py @router.get("/{user_id}", response_model=UserResponseV1) async def get_user_v1(user_id: str, service: UserService = Depends()): user = await service.get_user(user_id) return UserResponseV1(id=user.id, name=user.full_name) # v2/routes/users.py @router.get("/{user_id}", response_model=UserResponseV2) async def get_user_v2(user_id: str, service: UserService = Depends()): user = await service.get_user(user_id) return UserResponseV2( id=user.id, first_name=user.first_name, last_name=user.last_name, full_name=f"{user.first_name} {user.last_name}", ) ``` **Strategy Comparison:** | Strategy | Example | Pros | Cons | |----------|---------|------|------| | URL Path | `/api/v1/users` | Simple, visible, cacheable | URL pollution | | Header | `X-API-Version: 1` | Clean URLs | Hidden, harder to test | | Query Param | `?version=1` | Easy testing | Messy, cache issues | | Content-Type | `Accept: application/vnd.api.v1+json` | RESTful | Complex | **Incorrect — Versioned services:** ```python # Services should be version-agnostic class UserServiceV1: async def get_user(self, id: str): ... class UserServiceV2: async def get_user(self, id: str): ... ``` **Correct — Version-agnostic services:** ```python # Single service, version handled in response schemas class UserService: async def get_user(self, id: str) -> User: return await self.repo.get_by_id(id) # v1/routes/users.py returns UserResponseV1 # v2/routes/users.py returns UserResponseV2 ``` **Key rules:** - Always start with `/api/v1/` even if no v2 is planned - Keep services version-agnostic; only schemas and routes are versioned - Use schema inheritance for shared fields across versions - Support max 2-3 concurrent versions - Never version internal implementation (services, repositories) --- ## References (7) ### Graphql Api # GraphQL API Design ## Schema Design Principles ### Nullable by Default ```graphql type User { id: ID! # Non-null (required) email: String! # Non-null name: String # Nullable (optional) avatar: String # Nullable } ``` ### Use Connections for Lists ```graphql type Query { users(first: Int, after: String): UserConnection! } type UserConnection { edges: [UserEdge!]! pageInfo: PageInfo! totalCount: Int! } type UserEdge { node: User! cursor: String! } type PageInfo { hasNextPage: Boolean! hasPreviousPage: Boolean! startCursor: String endCursor: String } ``` ### Input Types for Mutations ```graphql input CreateUserInput { email: String! name: String! role: UserRole! } type Mutation { createUser(input: CreateUserInput!): CreateUserPayload! } type CreateUserPayload { user: User! errors: [UserError!] } type UserError { field: String! message: String! code: String! } ``` ## Query Design **Fetch single resource:** ```graphql query GetUser { user(id: "123") { id name email posts { id title } } } ``` **Fetch list with filters:** ```graphql query GetUsers { users( first: 10 after: "cursor123" filter: { role: DEVELOPER, status: ACTIVE } ) { edges { node { id name email } } pageInfo { hasNextPage endCursor } } } ``` ## Error Handling **Field-Level Errors:** ```graphql type CreateUserPayload { user: User errors: [UserError!] } ``` **Response:** ```json { "data": { "createUser": { "user": null, "errors": [ { "field": "email", "message": "Email is already taken", "code": "DUPLICATE_EMAIL" } ] } } } ``` ### Ork Delta # ork delta: api-design OrchestKit-specific decisions rescued during the wrap-plus-delta thinning of src/skills/api-design (2026-07-31). Vendor and spec tutorials that used to live beside these entries were deleted; the "Upstream coverage (do not restate)" table in SKILL.md points at their first-party sources. Only rules with a house decision or a scar behind them live here. ## Put problem type URIs on your own API domain under /problems/ Why: House convention carried since the error-handling-rfc9457 skill was consolidated into api-design v2.0.0 (metadata.json, February 2026). OrchestKit examples standardize on `https://api.orchestkit.dev/problems/<kebab-slug>` so error `type` URIs stay stable, documentable at their URL, and greppable across services (see examples/fastapi-problem-details.md for the full registry in use). Upstream: RFC 9457 Problem Details, https://www.rfc-editor.org/rfc/rfc9457.html ## Raise typed Problem exceptions, never bare HTTPException Why: House exception vocabulary (ProblemException base plus ResourceNotFoundError, ValidationError, ConflictError, RateLimitError, AuthenticationError, AuthorizationError, each carrying machine-readable extension members such as `resource_id` and `retry_after`) fixed at the v2.0.0 consolidation (February 2026). The agent-facing extensions from #1067 build on these same classes, and SKILL.md's quick start plus examples/fastapi-problem-details.md depend on them; ad-hoc HTTPException payloads broke error-format consistency across endpoints, which is what the consolidation was fixing. Upstream: FastAPI custom exception handlers, https://fastapi.tiangolo.com/tutorial/handling-errors/ ## Hold the deprecation window: 3 months notice, 6 months sunset, current + 1 supported Why: House lifecycle policy set when the api-versioning skill was consolidated into api-design v2.0.0 (February 2026): deprecation notice at least 3 months before sunset, sunset 6 months after deprecation, support latest stable plus 1 previous version, and never more than 2-3 concurrent versions. These numbers are OrchestKit policy, not spec requirements; the specs only define the header mechanics. The policy is also mirrored in SKILL.md Key Decisions so it survives file-level thinning. Upstream: RFC 8594 (Sunset header), https://www.rfc-editor.org/rfc/rfc8594.html and RFC 9745 (Deprecation header), https://www.rfc-editor.org/rfc/rfc9745.html ### Payload Vs Sanity # Payload vs Sanity — CMS Comparison Detailed comparison and decision matrix for choosing between Payload CMS 3.0, Sanity, Strapi, and WordPress. ## Feature Comparison | Feature | Payload 3.0 | Sanity v3 | Strapi v5 | WordPress | |---------|-------------|-----------|-----------|-----------| | **Language** | TypeScript | TypeScript + GROQ | JavaScript/TS | PHP | | **Framework** | Built on Next.js | React (studio) | Koa.js | Monolithic | | **Hosting** | Self-hosted | Hosted API + self-hosted studio | Self-hosted | Self/hosted | | **Database** | MongoDB or Postgres | Hosted (proprietary) | SQLite/Postgres/MySQL | MySQL | | **Auth** | Built-in (JWT + cookies) | Hosted or custom | Built-in (JWT) | Built-in (sessions) | | **API** | REST + GraphQL auto-generated | GROQ + GraphQL | REST + GraphQL | REST + GraphQL (plugin) | | **Rich Text** | Lexical (built-in) | Portable Text | CKEditor/custom | Gutenberg | | **Admin UI** | React + Next.js | React (Sanity Studio) | React | PHP + React (Gutenberg) | | **Type Safety** | Config IS the schema | Schema + codegen | Schema + codegen | None natively | | **Plugins** | npm packages | npm packages | npm marketplace | Plugin ecosystem (massive) | | **License** | MIT (open source) | Freemium (hosted) | MIT with EE features | GPLv2 | | **Live Preview** | Built-in | Built-in | Via plugin | Theme preview | | **Versioning** | Built-in per collection | Built-in | Via plugin | Built-in (revisions) | ## Cost Comparison | Tier | Payload | Sanity | Strapi | |------|---------|--------|--------| | **Free** | Unlimited (self-host) | 100K API requests/mo, 3 users | Unlimited (self-host) | | **Team** | Payload Cloud ($25/mo) | $99/mo (500K requests) | $29/mo (gold support) | | **Enterprise** | Custom | Custom | Custom | Payload is fully open source — cost is infrastructure only. Sanity's cost scales with API usage. ## Decision Matrix ### Choose Payload When: - Building a **Next.js application** — Payload runs inside your Next.js app - You want **full ownership** of data and infrastructure - Your team is **TypeScript-first** — config-as-code is natural - You need **custom access control** beyond simple roles - Self-hosting is acceptable or preferred - You want **one deployment** (CMS + frontend in same app) ### Choose Sanity When: - **Content editors** are primary users, not developers - You need **real-time collaborative editing** (Google Docs-style) - Your content is consumed by **multiple frontends** (web, mobile, IoT) - You want a **hosted API** with no infrastructure management - **GROQ** query language fits your content querying needs - Editorial workflow and content scheduling are critical ### Choose Strapi When: - You need a **quick admin panel** with minimal configuration - Your team prefers a **GUI-first** content modeling approach - You want a **marketplace** of pre-built plugins - The project is a **prototype or MVP** that may change CMS later - You need **multi-database support** (SQLite for dev, Postgres for prod) ### Choose WordPress When: - **Non-technical editors** need to manage content independently - You need the **largest plugin ecosystem** (100K+ plugins) - SEO tooling (Yoast, RankMath) is a core requirement - Budget for development is limited — large talent pool - Content is primarily **blog/marketing pages** ## Migration Considerations ### From Sanity to Payload 1. Export content via GROQ: `*[_type == "post"]` 2. Map Portable Text to Lexical rich text format 3. Recreate schemas as Payload collection configs 4. Migrate assets from Sanity CDN to local/S3 storage 5. Rebuild GROQ queries as Payload `where` clauses ### From Strapi to Payload 1. Export via Strapi REST API 2. Map Strapi content types to Payload collections 1:1 3. Convert Strapi lifecycle hooks to Payload hooks 4. Migrate media from Strapi uploads to Payload upload collections 5. Replace Strapi custom controllers with Payload custom endpoints ### From WordPress to Payload 1. Export via WP REST API (`/wp-json/wp/v2/posts`) 2. Convert ACF/custom fields to Payload field configs 3. Map WordPress taxonomies to Payload relationship fields 4. Migrate media library to Payload upload collection 5. Convert WordPress template hierarchy to Next.js layouts ## Architecture Comparison ``` Payload 3.0: Sanity: ┌─────────────────────┐ ┌──────────────┐ ┌──────────────┐ │ Your Next.js App │ │ Sanity Studio │ │ Your App │ │ ┌───────────────┐ │ │ (React SPA) │ │ (any framework)│ │ │ Payload CMS │ │ └──────┬───────┘ └──────┬───────┘ │ │ (embedded) │ │ │ │ │ └───────┬───────┘ │ ▼ ▼ │ │ │ ┌──────────────┐ ┌──────────────┐ │ ▼ │ │ Sanity API │ │ Sanity API │ │ ┌───────────────┐ │ │ (hosted) │ │ (hosted) │ │ │ MongoDB/PG │ │ └──────────────┘ └──────────────┘ │ └───────────────┘ │ └─────────────────────┘ Single hosted API, multiple consumers Single deployment, full control ``` ## When NOT to Use a Headless CMS - **Static content** that rarely changes — use Markdown + static site generator - **Application data** (user profiles, orders, analytics) — use a database directly - **Real-time data** (chat, live feeds) — use purpose-built real-time tools - Content that only developers edit — YAML/JSON config files may suffice ### Rest Patterns # RESTful API Design Patterns Comprehensive guide to RESTful API design patterns including resource modeling, HTTP methods, status codes, versioning, pagination, filtering, and error handling. ## Resource Modeling ### Naming Conventions **Use plural nouns for collections:** ``` ✅ GET /api/v1/analyses ✅ GET /api/v1/artifacts ✅ GET /api/v1/users ❌ GET /api/v1/analysis ❌ GET /api/v1/getArtifact ``` **Hierarchical relationships:** ``` ✅ GET /api/v1/analyses/{analysis_id}/artifact ✅ GET /api/v1/teams/{team_id}/members ✅ POST /api/v1/projects/{project_id}/tasks ❌ GET /api/v1/artifact?analysis_id={id} # Query param for relationship ❌ GET /api/v1/analysis_artifact/{id} # Flat structure ``` **Use kebab-case for multi-word resources:** ``` ✅ /api/v1/shopping-carts ✅ /api/v1/user-preferences ✅ /api/v1/order-items ❌ /api/v1/shoppingCarts (camelCase) ❌ /api/v1/shopping_carts (snake_case in URL) ``` ### HTTP Methods (CRUD Operations) | Method | Purpose | Idempotent | Safe | Response | Example | |--------|---------|------------|------|----------|---------| | **GET** | Retrieve resource(s) | ✅ | ✅ | 200 OK | `GET /analyses/123` | | **POST** | Create resource | ❌ | ❌ | 201 Created | `POST /analyses` | | **PUT** | Replace entire resource | ✅ | ❌ | 200 OK | `PUT /analyses/123` | | **PATCH** | Partial update | ⚠️ | ❌ | 200 OK | `PATCH /analyses/123` | | **DELETE** | Remove resource | ✅ | ❌ | 204 No Content | `DELETE /analyses/123` | | **HEAD** | Metadata only | ✅ | ✅ | 200 OK | `HEAD /analyses/123` | | **OPTIONS** | Allowed methods | ✅ | ✅ | 200 OK | `OPTIONS /analyses` | **Idempotency Note**: PATCH can be designed to be idempotent by using absolute values instead of relative operations. ### HTTP Status Codes #### Success (2xx) **200 OK** - Successful GET, PUT, PATCH, DELETE with response body ```python @router.get("/analyses/{analysis_id}") async def get_analysis(analysis_id: uuid.UUID) -> AnalysisResponse: return AnalysisResponse(...) # 200 OK ``` **201 Created** - Successful POST, include `Location` header ```python @router.post("/analyses", status_code=status.HTTP_201_CREATED) async def create_analysis(request: AnalyzeRequest) -> AnalyzeCreateResponse: # Include SSE endpoint in response return AnalyzeCreateResponse( analysis_id=str(analysis_uuid), sse_endpoint=f"/api/v1/analyze/{analysis_uuid}/stream" ) ``` **202 Accepted** - Request accepted, processing asynchronously ```python @router.post("/long-running-task", status_code=status.HTTP_202_ACCEPTED) async def start_task() -> TaskStatusResponse: # Start background task return TaskStatusResponse( task_id="...", status="pending", status_url="/tasks/123/status" ) ``` **204 No Content** - Successful DELETE or PUT with no response body ```python @router.delete("/analyses/{analysis_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_analysis(analysis_id: uuid.UUID) -> None: await repo.delete(analysis_id) ``` #### Client Errors (4xx) **400 Bad Request** - Invalid request syntax or malformed parameters ```json { "error": { "code": "INVALID_REQUEST", "message": "Request body is not valid JSON", "timestamp": "2025-12-21T10:30:00Z" } } ``` **401 Unauthorized** - Missing or invalid authentication ```json { "error": { "code": "UNAUTHORIZED", "message": "Missing or invalid authentication token", "timestamp": "2025-12-21T10:30:00Z" } } ``` **403 Forbidden** - Authenticated but not authorized ```json { "error": { "code": "FORBIDDEN", "message": "You do not have permission to access this resource", "timestamp": "2025-12-21T10:30:00Z" } } ``` **404 Not Found** - Resource doesn't exist ```python @router.get("/artifacts/{artifact_id}") async def get_artifact(artifact_id: uuid.UUID) -> ArtifactResponse: artifact = await repo.get_artifact_by_id(artifact_id) if not artifact: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Artifact {artifact_id} not found" ) ``` **422 Unprocessable Entity** - Validation failed ```python try: content_type = detect_content_type(url_str) except ContentTypeError as e: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Invalid URL format: {e!s}" ) from e ``` **429 Too Many Requests** - Rate limit exceeded ```json HTTP/1.1 429 Too Many Requests Retry-After: 3600 X-RateLimit-Limit: 1000 X-RateLimit-Remaining: 0 X-RateLimit-Reset: 1703163600 { "error": { "code": "RATE_LIMIT_EXCEEDED", "message": "API rate limit exceeded. Try again in 1 hour.", "retry_after": 3600 } } ``` #### Server Errors (5xx) **500 Internal Server Error** - Generic server error ```python except Exception as e: logger.error( "analysis_creation_failed", error=str(e), exc_info=True ) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create analysis record" ) from e ``` **502 Bad Gateway** - Upstream service error **503 Service Unavailable** - Temporary unavailability (maintenance) **504 Gateway Timeout** - Upstream timeout ## API Versioning ### Strategy 1: URI Versioning (Recommended for Public APIs) **OrchestKit uses this approach:** ```python # app/core/config.py API_V1_PREFIX = "/api/v1" # app/main.py app.include_router( analysis_router, prefix=f"{settings.API_V1_PREFIX}/analyze" ) ``` **URL structure:** ``` /api/v1/analyses /api/v2/analyses # New version with breaking changes ``` **Pros:** - Clear and visible in URLs - Easy to test and debug - Cache-friendly - Can route different versions to different servers **Cons:** - Verbose URLs - Need to maintain multiple codebases ### Strategy 2: Header Versioning ``` GET /api/analyses Accept: application/vnd.orchestkit.v2+json API-Version: v2 ``` **Pros:** - Clean URLs - RESTful purist approach **Cons:** - Not visible in browser - Harder to test manually - Need custom headers ### Strategy 3: Query Parameter (Avoid) ``` GET /api/analyses?version=2 ``` **Cons:** - Mixes with business logic parameters - Can be forgotten - Not cache-friendly ## Pagination ### Cursor-Based Pagination (Recommended for Large Datasets) **Best for**: Real-time data, infinite scroll, datasets that change frequently ```python @router.get("/analyses") async def list_analyses( cursor: str | None = None, limit: int = Query(default=20, le=100) ) -> PaginatedResponse: results = await repo.get_paginated(cursor=cursor, limit=limit) return { "data": results, "pagination": { "next_cursor": encode_cursor(results[-1].id) if results else None, "has_more": len(results) == limit } } ``` **Response:** ```json { "data": [...], "pagination": { "next_cursor": "eyJpZCI6MTIzfQ", "has_more": true } } ``` **Client usage:** ```javascript // First page const page1 = await fetch('/api/v1/analyses?limit=20') const { data, pagination } = await page1.json() // Next page if (pagination.has_more) { const page2 = await fetch(`/api/v1/analyses?cursor=${pagination.next_cursor}&limit=20`) } ``` ### Offset-Based Pagination (For Known Bounds) **Best for**: Admin panels, small datasets, "jump to page N" UX ```python @router.get("/analyses") async def list_analyses( page: int = Query(default=1, ge=1), per_page: int = Query(default=20, le=100) ) -> PaginatedResponse: offset = (page - 1) * per_page results, total = await repo.get_paginated(offset=offset, limit=per_page) return { "data": results, "pagination": { "page": page, "per_page": per_page, "total": total, "total_pages": (total + per_page - 1) // per_page } } ``` **Response:** ```json { "data": [...], "pagination": { "page": 2, "per_page": 20, "total": 487, "total_pages": 25 } } ``` ## Filtering and Sorting ### Query Parameter Filtering ```python @router.get("/analyses") async def list_analyses( status: str | None = None, content_type: str | None = None, created_after: datetime | None = None, created_before: datetime | None = None ) -> list[AnalysisResponse]: filters = {} if status: filters["status"] = status if content_type: filters["content_type"] = content_type # ... return await repo.find_all(filters=filters) ``` **Usage:** ``` GET /api/v1/analyses?status=completed&content_type=article GET /api/v1/analyses?created_after=2025-01-01&created_before=2025-12-31 ``` ### Sorting ```python @router.get("/analyses") async def list_analyses( sort: str = Query(default="-created_at") ) -> list[AnalysisResponse]: # Parse sort parameter: "-created_at" -> ("created_at", "desc") direction = "desc" if sort.startswith("-") else "asc" field = sort.lstrip("-") return await repo.find_all( order_by=field, direction=direction ) ``` **Usage:** ``` GET /api/v1/analyses?sort=-created_at # Newest first GET /api/v1/analyses?sort=title # Alphabetical GET /api/v1/analyses?sort=-status,title # Multiple fields ``` ### Field Selection (Sparse Fieldsets) ```python @router.get("/analyses") async def list_analyses( fields: str | None = None ) -> list[dict[str, Any]]: selected_fields = fields.split(",") if fields else None results = await repo.find_all() if selected_fields: return [ {k: v for k, v in item.model_dump().items() if k in selected_fields} for item in results ] return results ``` **Usage:** ``` GET /api/v1/analyses?fields=id,title,status ``` ## Error Response Format ### Standard Error Structure ```python # app/api/schemas/errors.py from pydantic import BaseModel, ConfigDict class ErrorDetail(BaseModel): field: str message: str code: str class ErrorResponse(BaseModel): error: dict[str, Any] model_config = ConfigDict( json_schema_extra={ "example": { "error": { "code": "VALIDATION_ERROR", "message": "Request validation failed", "details": [ { "field": "url", "message": "Invalid URL format", "code": "INVALID_URL" } ], "timestamp": "2025-12-21T10:30:00Z", "request_id": "req_abc123" } } } ) ``` ### FastAPI Exception Handlers ```python # app/main.py @app.exception_handler(HTTPException) async def http_exception_handler(request: Request, exc: HTTPException): return JSONResponse( status_code=exc.status_code, content={ "error": { "code": exc.status_code, "message": exc.detail, "timestamp": datetime.now(UTC).isoformat(), "path": request.url.path } } ) @app.exception_handler(ValidationError) async def validation_exception_handler(request: Request, exc: ValidationError): errors = [] for error in exc.errors(): errors.append({ "field": ".".join(str(x) for x in error["loc"]), "message": error["msg"], "code": error["type"] }) return JSONResponse( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content={ "error": { "code": "VALIDATION_ERROR", "message": "Request validation failed", "details": errors, "timestamp": datetime.now(UTC).isoformat() } } ) ``` ## Rate Limiting ### Response Headers ```python from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) @router.get("/analyses") @limiter.limit("100/minute") async def list_analyses(request: Request) -> list[AnalysisResponse]: # Rate limited to 100 requests per minute pass ``` **Response headers:** ``` HTTP/1.1 200 OK X-RateLimit-Limit: 100 X-RateLimit-Remaining: 87 X-RateLimit-Reset: 1703163600 ``` **When exceeded:** ``` HTTP/1.1 429 Too Many Requests Retry-After: 60 X-RateLimit-Limit: 100 X-RateLimit-Remaining: 0 X-RateLimit-Reset: 1703163600 { "error": { "code": "RATE_LIMIT_EXCEEDED", "message": "Too many requests. Please try again in 60 seconds.", "retry_after": 60, "timestamp": "2025-12-21T10:30:00Z" } } ``` ## Best Practices ### 1. Always Return Consistent Response Format ```python # Good: Consistent structure { "data": {...}, "metadata": {...} } # Bad: Inconsistent structure {...} # Sometimes flat object {"results": [...]} # Sometimes wrapped ``` ### 2. Use Pydantic for Request/Response Validation ```python from pydantic import BaseModel, HttpUrl, Field class AnalyzeRequest(BaseModel): url: HttpUrl analysis_id: str | None = None skill_level: str = Field(default="beginner", pattern="^(beginner|intermediate|advanced)$") ``` ### 3. Include Metadata in Responses ```python { "analysis_id": "123", "url": "https://example.com", "created_at": "2025-12-21T10:30:00Z", "updated_at": "2025-12-21T11:00:00Z" } ``` ### 4. Use OpenAPI Documentation ```python @router.get( "/analyses/{analysis_id}", responses={ 404: {"model": ErrorResponse, "description": "Analysis not found"}, 500: {"model": ErrorResponse, "description": "Internal server error"} }, summary="Get analysis details", description="Retrieve detailed information about a specific analysis including status and artifacts" ) async def get_analysis( analysis_id: Annotated[uuid.UUID, Path(description="Analysis UUID")] ) -> AnalysisResponse: ... ``` ### 5. Handle Edge Cases ```python # Empty collections: Return empty array, not null {"data": []} # ✅ {"data": null} # ❌ # Deleted resources: Return 404, not null # ❌ {"data": null} # ✅ 404 Not Found # Null fields: Be explicit { "title": null, # ✅ Explicitly null "description": "" # ✅ Empty string if required } ``` ## Related Files - See `assets/openapi-template.yaml` for full OpenAPI specification example - See `examples/orchestkit-api-design.md` for OrchestKit-specific patterns - See SKILL.md for GraphQL and gRPC patterns ### Telegram Bot Api # Telegram Bot API Reference for building Telegram bots with webhooks, commands, and interactive keyboards. ## Bot Creation 1. Open Telegram, search for `@BotFather` 2. Send `/newbot`, follow prompts 3. Save the bot token (`123456:ABC-DEF...`) ## Webhook Setup ```bash # Set webhook curl -X POST "https://api.telegram.org/bot