⚒️ 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---2id: FORGE-fastapi-api-builder3name: forge-fastapi-api-builder4version: 1.0.0-2026.07.175description: "FastAPI API builder for organ bridge middleware and federation REST endpoints."6owner: A-FORGE7risk_tier: medium8floor_scope: ['F1', 'F4', 'F12']9autonomy_tier: T110capability_tier: fed-agent-subagent11ecology_state: WARM12---1314# ⚒️ FastAPI API Builder — Organ Bridge Middleware1516> **DITEMPA BUKAN DIBERI** — Forged, Not Given.1718## Purpose19Build and maintain FastAPI/Node API endpoints for `/api/observatory/v1/*` and organ bridge surfaces: middleware stack, CORS, error handling, health probes, MCP proxy endpoints.2021## When to Use22- Creating new API endpoints in arifOS, GEOX, WEALTH, WELL FastAPI servers23- Adding middleware — auth, CORS, request logging, rate limiting24- Error handling patterns — structured error responses, exception handlers25- MCP-to-HTTP bridge endpoints for observatory consumption2627## When NOT to Use28- Frontend routes — use `nextjs-mastery` or `react-spa-discipline`29- Database schema design — use `postgres-schema-design`30- Deployment/Docker — use `cicd-docker-deploy`3132## Constitutional Floor Alignment3334| Floor | Application |35|-------|-------------|36| F1 AMANAH | Version all breaking API changes (`/v1/`, `/v2/`); never mutate in place |37| F2 TRUTH | Response schemas must match documented contracts; no undocumented fields |38| F4 CLARITY | One endpoint = one responsibility; no mega-endpoints |39| F11 AUDIT | Every request logged with actor, intent, timestamp |40| F12 INJECTION | All inputs sanitized; never trust request body without validation |4142## Commands & Patterns4344```python45# FastAPI organ endpoint pattern46from fastapi import APIRouter, HTTPException47from pydantic import BaseModel4849router = APIRouter(prefix="/api/observatory/v1")5051class HealthResponse(BaseModel):52 status: str53 version: str54 uptime: float5556@router.get("/health", response_model=HealthResponse)57async def health():58 return {"status": "ok", "version": "1.0.0", "uptime": time.monotonic()}5960# CORS middleware61app.add_middleware(CORSMiddleware, allow_origins=["https://arifos.arif-fazil.com"])6263# Structured error handler64@app.exception_handler(AppError)65async def app_error_handler(request, exc):66 return JSONResponse(status_code=exc.code, content={"error": exc.message, "trace": exc.trace_id})67```6869## Refusal Surface70- ❌ Endpoints that mutate without lease/session auth71- ❌ Returning raw database models as response (always use Pydantic schema)72- ❌ Skipping input validation on POST/PUT/PATCH endpoints73- ❌ Hardcoded organ ports in API routes — use env config74- ❌ Sync endpoints for I/O-bound operations (must be async)