⚒️ FastAPI API Builder — Organ Bridge Middleware
DITEMPA BUKAN DIBERI — Forged, Not Given.
Purpose
Build and maintain FastAPI/Node API endpoints for /api/observatory/v1/* and organ bridge surfaces: middleware stack, CORS, error handling, health probes, MCP proxy endpoints.
When to Use
- Creating new API endpoints in arifOS, GEOX, WEALTH, WELL FastAPI servers
- Adding middleware — auth, CORS, request logging, rate limiting
- Error handling patterns — structured error responses, exception handlers
- MCP-to-HTTP bridge endpoints for observatory consumption
When NOT to Use
- Frontend routes — use
nextjs-mastery or react-spa-discipline
- Database schema design — use
postgres-schema-design
- Deployment/Docker — use
cicd-docker-deploy
Constitutional Floor Alignment
| Floor |
Application |
| F1 AMANAH |
Version all breaking API changes (/v1/, /v2/); never mutate in place |
| F2 TRUTH |
Response schemas must match documented contracts; no undocumented fields |
| F4 CLARITY |
One endpoint = one responsibility; no mega-endpoints |
| F11 AUDIT |
Every request logged with actor, intent, timestamp |
| F12 INJECTION |
All inputs sanitized; never trust request body without validation |
Commands & Patterns
# FastAPI organ endpoint pattern
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
router = APIRouter(prefix="/api/observatory/v1")
class HealthResponse(BaseModel):
status: str
version: str
uptime: float
@router.get("/health", response_model=HealthResponse)
async def health():
return {"status": "ok", "version": "1.0.0", "uptime": time.monotonic()}
# CORS middleware
app.add_middleware(CORSMiddleware, allow_origins=["https://arifos.arif-fazil.com"])
# Structured error handler
@app.exception_handler(AppError)
async def app_error_handler(request, exc):
return JSONResponse(status_code=exc.code, content={"error": exc.message, "trace": exc.trace_id})
Refusal Surface
- ❌ Endpoints that mutate without lease/session auth
- ❌ Returning raw database models as response (always use Pydantic schema)
- ❌ Skipping input validation on POST/PUT/PATCH endpoints
- ❌ Hardcoded organ ports in API routes — use env config
- ❌ Sync endpoints for I/O-bound operations (must be async)
1---2name: forge-fastapi-api-builder3description: FastAPI API builder for organ bridge middleware and federation REST endpoints.4---5# ⚒️ FastAPI API Builder — Organ Bridge Middleware67> **DITEMPA BUKAN DIBERI** — Forged, Not Given.89## Purpose10Build and maintain FastAPI/Node API endpoints for `/api/observatory/v1/*` and organ bridge surfaces: middleware stack, CORS, error handling, health probes, MCP proxy endpoints.1112## When to Use13- Creating new API endpoints in arifOS, GEOX, WEALTH, WELL FastAPI servers14- Adding middleware — auth, CORS, request logging, rate limiting15- Error handling patterns — structured error responses, exception handlers16- MCP-to-HTTP bridge endpoints for observatory consumption1718## When NOT to Use19- Frontend routes — use `nextjs-mastery` or `react-spa-discipline`20- Database schema design — use `postgres-schema-design`21- Deployment/Docker — use `cicd-docker-deploy`2223## Constitutional Floor Alignment2425| Floor | Application |26|-------|-------------|27| F1 AMANAH | Version all breaking API changes (`/v1/`, `/v2/`); never mutate in place |28| F2 TRUTH | Response schemas must match documented contracts; no undocumented fields |29| F4 CLARITY | One endpoint = one responsibility; no mega-endpoints |30| F11 AUDIT | Every request logged with actor, intent, timestamp |31| F12 INJECTION | All inputs sanitized; never trust request body without validation |3233## Commands & Patterns3435```python36# FastAPI organ endpoint pattern37from fastapi import APIRouter, HTTPException38from pydantic import BaseModel3940router = APIRouter(prefix="/api/observatory/v1")4142class HealthResponse(BaseModel):43 status: str44 version: str45 uptime: float4647@router.get("/health", response_model=HealthResponse)48async def health():49 return {"status": "ok", "version": "1.0.0", "uptime": time.monotonic()}5051# CORS middleware52app.add_middleware(CORSMiddleware, allow_origins=["https://arifos.arif-fazil.com"])5354# Structured error handler55@app.exception_handler(AppError)56async def app_error_handler(request, exc):57 return JSONResponse(status_code=exc.code, content={"error": exc.message, "trace": exc.trace_id})58```5960## Refusal Surface61- ❌ Endpoints that mutate without lease/session auth62- ❌ Returning raw database models as response (always use Pydantic schema)63- ❌ Skipping input validation on POST/PUT/PATCH endpoints64- ❌ Hardcoded organ ports in API routes — use env config65- ❌ Sync endpoints for I/O-bound operations (must be async)