Scaffolding Fastapi Dapr

FastAPI + Dapr Backend

Build production-grade FastAPI backends with SQLModel, Dapr integration, and JWT authentication.

Quick Start

# Project setup
uv init backend && cd backend
uv add fastapi sqlmodel pydantic httpx python-jose uvicorn

# Development
uv run uvicorn main:app --reload --port 8000

# With Dapr sidecar
dapr run --app-id myapp --app-port 8000 -- uvicorn main:app

---

FastAPI Core Patterns

1. SQLModel Schema (Database + API)

from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional, Literal

class TaskBase(SQLModel):
    title: str = Field(max_length=200, index=True)
    status: Literal["pending", "in_progress", "completed"] = "pending"

class Task(TaskBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.now)

class TaskCreate(TaskBase):
    pass

class TaskRead(TaskBase):
    id: int
    created_at: datetime

2. Async Database Setup

from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
import os

DATABASE_URL = os.getenv("DATABASE_URL").replace("postgresql://", "postgresql+asyncpg://")
engine = create_async_engine(DATABASE_URL)

async def get_session() -> AsyncSession:
    async with AsyncSession(engine) as session:
        yield session

3. CRUD Endpoints

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import select

app = FastAPI()

@app.post("/tasks", response_model=TaskRead, status_code=201)
async def create_task(task: TaskCreate, session: AsyncSession = Depends(get_session)):
    db_task = Task.model_validate(task)
    session.add(db_task)
    await session.commit()
    await session.refresh(db_task)
    return db_task

@app.get("/tasks/{task_id}", response_model=TaskRead)
async def get_task(task_id: int, session: AsyncSession = Depends(get_session)):
    task = await session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Not found")
    return task

@app.patch("/tasks/{task_id}", response_model=TaskRead)
async def update_task(task_id: int, update: TaskUpdate, session: AsyncSession = Depends(get_session)):
    task = await session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Not found")
    update_data = update.model_dump(exclude_unset=True)
    task.sqlmodel_update(update_data)
    session.add(task)
    await session.commit()
    await session.refresh(task)
    return task

4. JWT/JWKS Authentication

from jose import jwt
import httpx

JWKS_URL = f"{SSO_URL}/.well-known/jwks.json"

async def get_current_user(authorization: str = Header()):
    token = authorization.replace("Bearer ", "")
    async with httpx.AsyncClient() as client:
        jwks = (await client.get(JWKS_URL)).json()
    payload = jwt.decode(token, jwks, algorithms=["RS256"])
    return payload

@app.get("/protected")
async def protected_route(user = Depends(get_current_user)):
    return {"user": user["sub"]}

See references/fastapi-patterns.md for audit logging, pagination, and OpenAPI configuration.

---

Dapr Integration Patterns

1. Pub/Sub Subscription

from fastapi import APIRouter, Request

router = APIRouter(prefix="/dapr", tags=["Dapr"])

@router.get("/subscribe")
async def subscribe():
    """Dapr calls this to discover subscriptions."""
    return [{
        "pubsubname": "pubsub",
        "topic": "task-created",
        "route": "/dapr/task-created"
    }]

@router.post("/task-created")
async def handle_task_created(request: Request, session: AsyncSession = Depends(get_session)):
    # CloudEvent wrapper - data is nested
    event = await request.json()
    task_data = event.get("data", event)  # Handle both wrapped and unwrapped

    # Process event
    task = Task.model_validate(task_data)
    session.add(task)
    await session.commit()
    return {"status": "processed"}

2. Publishing Events

import httpx

DAPR_URL = "http://localhost:3500"

async def publish_event(topic: str, data: dict):
    async with httpx.AsyncClient() as client:
        await client.post(
            f"{DAPR_URL}/v1.0/publish/pubsub/{topic}",
            json=data,
            headers={"Content-Type": "application/json"}
        )

3. Scheduled Jobs

# Schedule a job via Dapr Jobs API (alpha)
async def schedule_job(name: str, schedule: str, callback_url: str, data: dict):
    async with httpx.AsyncClient() as client:
        await client.post(
            f"{DAPR_URL}/v1.0-alpha1/jobs/{name}",
            json={
                "schedule": schedule,  # "@every 5m" or "0 */5 * * * *"
                "data": data,
            },
            headers={"dapr-app-callback-url": callback_url}
        )

# Job callback endpoint
@app.post("/jobs/process")
async def process_job(request: Request):
    job_data = await request.json()
    # Handle job execution
    return {"status": "completed"}

See references/dapr-patterns.md for state management and advanced patterns.

---

Production Patterns

Structured Logging

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ]
)
log = structlog.get_logger()
log.info("task_created", task_id=task.id, user_id=user["sub"])

Repository + Service Pattern

# Repository: data access only
class TaskRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def create(self, task: TaskCreate) -> Task:
        db_task = Task.model_validate(task)
        self.session.add(db_task)
        await self.session.commit()
        return db_task

# Service: business logic
class TaskService:
    def __init__(self, repo: TaskRepository):
        self.repo = repo

    async def create_task(self, task: TaskCreate, user_id: str) -> Task:
        # Business logic here
        return await self.repo.create(task)

# Dependency injection
def get_task_service(session: AsyncSession = Depends(get_session)):
    return TaskService(TaskRepository(session))

Async Testing

@pytest.fixture
async def client(session):
    app.dependency_overrides[get_session] = lambda: session
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

@pytest.mark.anyio
async def test_create_task(client: AsyncClient):
    response = await client.post("/tasks", json={"title": "Test"})
    assert response.status_code == 201

See references/production-testing.md for full patterns.

---

Project Structure

backend/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI app
│   ├── database.py       # Async engine + session
│   ├── models/           # SQLModel schemas
│   ├── routers/          # API routes
│   ├── repositories/     # Data access layer
│   ├── services/         # Business logic
│   └── dapr/             # Dapr handlers
├── tests/
│   ├── conftest.py       # Fixtures
│   └── test_*.py         # Test files
├── components/           # Dapr components (k8s)
│   ├── pubsub.yaml
│   └── statestore.yaml
└── pyproject.toml

---

Verification

Run: python3 scripts/verify.py

Expected: ✓ scaffolding-fastapi-dapr skill ready

If Verification Fails

1. Check: references/ folder has both pattern files 2. Stop and report if still failing

Related Skills

• configuring-better-auth - JWT/JWKS auth for API endpoints
• fetching-library-docs - FastAPI docs: --library-id /fastapi/fastapi --topic dependencies

References

• references/fastapi-patterns.md - Complete FastAPI backend patterns
• references/dapr-patterns.md - Dapr pub/sub, state, and jobs
• references/sqlmodel-patterns.md - SQLModel database patterns and migrations
• references/production-testing.md - Structured logging, DI, testing, versioning

Dapr Integration

Integrate Dapr sidecar for pub/sub messaging, state management, and scheduled jobs in Kubernetes environments.

When to Use

• Setting up Dapr pub/sub for event-driven microservices
• Scheduling jobs with Dapr Jobs API (v1.0-alpha1)
• Handling CloudEvent message formats
• Implementing subscription handlers in FastAPI
• Debugging Dapr integration issues

Quick Start

# Install Dapr CLI
curl -fsSL https://raw.githubusercontent.com/dapr/cli/master/install/install.sh | bash

# Initialize Dapr (local)
dapr init

# Run app with Dapr sidecar
dapr run --app-id myapp --app-port 8000 -- uvicorn main:app

Core Patterns

1. Pub/Sub Subscription Handler (FastAPI)

from fastapi import APIRouter, Request
from sqlmodel.ext.asyncio.session import AsyncSession

router = APIRouter(prefix="/dapr", tags=["Dapr"])

# Topics we subscribe to
SUBSCRIPTIONS = [
    {"pubsubname": "taskflow-pubsub", "topic": "task-events", "route": "/dapr/events/task-events"},
    {"pubsubname": "taskflow-pubsub", "topic": "reminders", "route": "/dapr/events/reminders"},
]

@router.get("/subscribe")
async def get_subscriptions() -> list[dict]:
    """Dapr calls this on startup to discover subscriptions."""
    return SUBSCRIPTIONS

2. CloudEvent Handling (CRITICAL)

Dapr wraps all pub/sub messages in CloudEvent format. You MUST unwrap it.

@router.post("/events/task-events")
async def handle_task_events(
    request: Request,
    session: AsyncSession = Depends(get_session),
) -> dict:
    try:
        # Step 1: Get raw CloudEvent
        raw_event = await request.json()

        # Step 2: ALWAYS unwrap CloudEvent "data" field
        # CloudEvent structure:
        # {
        #   "data": {              <-- Your payload is HERE
        #     "event_type": "task.created",
        #     "data": {...},
        #     "timestamp": "..."
        #   },
        #   "datacontenttype": "application/json",
        #   "id": "...",
        #   "pubsubname": "taskflow-pubsub",
        #   "source": "myapp",
        #   "topic": "task-events",
        #   ...
        # }
        event = raw_event.get("data", raw_event)  # Unwrap or use as-is

        # Step 3: Now access your payload
        event_type = event.get("event_type")  # "task.created"
        data = event.get("data", {})          # Your actual data

        # Process event...

        return {"status": "SUCCESS"}

    except Exception as e:
        logger.exception("Error handling event: %s", e)
        # Return SUCCESS to prevent Dapr retries for bad events
        return {"status": "SUCCESS"}

3. Publishing Events

import httpx

DAPR_HTTP_ENDPOINT = "http://localhost:3500"
PUBSUB_NAME = "taskflow-pubsub"

async def publish_event(
    topic: str,
    event_type: str,
    data: dict,
) -> bool:
    """Publish event to Dapr pub/sub."""
    url = f"{DAPR_HTTP_ENDPOINT}/v1.0/publish/{PUBSUB_NAME}/{topic}"

    payload = {
        "event_type": event_type,
        "data": data,
        "timestamp": datetime.utcnow().isoformat(),
    }

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.post(url, json=payload)
            return response.status_code == 204
    except Exception as e:
        logger.error("Failed to publish event: %s", e)
        return False

4. Dapr Jobs API (Scheduled Jobs)

CRITICAL: Dapr Jobs v1.0-alpha1 calls back to /job/{job_name} by default!

# Scheduling a job
async def schedule_job(
    job_name: str,
    due_time: datetime,
    data: dict,
    dapr_http_endpoint: str = "http://localhost:3500",
) -> bool:
    """Schedule a one-time Dapr job."""
    url = f"{dapr_http_endpoint}/v1.0-alpha1/jobs/{job_name}"

    payload = {
        "dueTime": due_time.strftime("%Y-%m-%dT%H:%M:%SZ"),  # RFC3339
        "data": data,
    }

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.post(url, json=payload)
            return response.status_code == 204
    except Exception as e:
        logger.error("Failed to schedule job: %s", e)
        return False

Handling the callback - Dapr calls /job/{job_name}, NOT a custom endpoint:

# WRONG - Dapr won't call this!
@router.post("/api/jobs/trigger")
async def handle_trigger(...):
    pass

# CORRECT - This is what Dapr actually calls
@router.post("/job/{job_name}")
async def handle_dapr_job_callback(
    job_name: str,
    request: Request,
    session: AsyncSession = Depends(get_session),
) -> dict:
    """Handle Dapr Jobs v1.0-alpha1 callback.

    Dapr calls POST /job/{job_name} when a scheduled job fires.
    """
    try:
        body = await request.json()
        job_data = body.get("data", body)  # Unwrap if needed

        task_id = job_data.get("task_id")
        job_type = job_data.get("type")

        logger.info("Job callback: job=%s, type=%s", job_name, job_type)

        if job_type == "reminder":
            return await handle_reminder(session, job_data)
        elif job_type == "spawn":
            return await handle_spawn(session, task_id)

        return {"status": "unknown_type"}

    except Exception as e:
        logger.exception("Error handling job %s: %s", job_name, e)
        return {"status": "error"}

5. Deleting Scheduled Jobs

async def delete_job(
    job_name: str,
    dapr_http_endpoint: str = "http://localhost:3500",
) -> bool:
    """Cancel a scheduled Dapr job."""
    url = f"{dapr_http_endpoint}/v1.0-alpha1/jobs/{job_name}"

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.delete(url)
            # 204 = deleted, 500 = not found (both OK)
            return response.status_code in (204, 500)
    except Exception as e:
        logger.error("Failed to delete job: %s", e)
        return False

Kubernetes/Helm Configuration

Dapr Pub/Sub Component (Redis)

# dapr-pubsub.yaml
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: taskflow-pubsub
  namespace: taskflow
spec:
  type: pubsub.redis
  version: v1
  metadata:
    - name: redisHost
      value: "redis:6379"
    - name: redisPassword
      secretKeyRef:
        name: redis-secret
        key: password
    - name: enableTLS
      value: "true"  # Required for Upstash

Dapr Annotations for Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: taskflow-api
spec:
  template:
    metadata:
      annotations:
        dapr.io/enabled: "true"
        dapr.io/app-id: "taskflow-api"
        dapr.io/app-port: "8000"
        dapr.io/enable-api-logging: "true"

Common Pitfalls

1. Not Unwrapping CloudEvent

# WRONG - event_type will be None!
event = await request.json()
event_type = event.get("event_type")  # None - it's nested in "data"

# CORRECT
raw_event = await request.json()
event = raw_event.get("data", raw_event)  # Unwrap CloudEvent
event_type = event.get("event_type")  # "task.created"

2. Wrong Job Callback URL

# WRONG - Dapr calls /job/{name}, not custom endpoints
@router.post("/api/jobs/trigger")  # Dapr won't call this!

# CORRECT
@router.post("/job/{job_name}")  # Dapr WILL call this

3. Forgetting to Return SUCCESS

# WRONG - Dapr will retry on errors
@router.post("/events/task-events")
async def handle(request: Request):
    try:
        # process...
        return {"status": "SUCCESS"}
    except Exception:
        raise  # Dapr will retry!

# CORRECT - Always return SUCCESS to stop retries
@router.post("/events/task-events")
async def handle(request: Request):
    try:
        # process...
    except Exception as e:
        logger.exception("Error: %s", e)
    return {"status": "SUCCESS"}  # Always, even on error

4. Using Wrong Dapr HTTP Port

# Local development
DAPR_HTTP_ENDPOINT = "http://localhost:3500"

# In Kubernetes (sidecar)
DAPR_HTTP_ENDPOINT = "http://localhost:3500"  # Same! Sidecar is localhost

Debugging

Check Dapr Sidecar Logs

# Kubernetes
kubectl logs deploy/myapp -c daprd -n mynamespace

# Look for:
# - "Scheduler stream connected" = Jobs API working
# - "HTTP API Called" = API calls to Dapr

Verify Subscriptions

# Call your subscribe endpoint
curl http://localhost:8000/dapr/subscribe

# Should return your subscriptions list

Test Pub/Sub Locally

# Publish test event
curl -X POST http://localhost:3500/v1.0/publish/taskflow-pubsub/task-events \
  -H "Content-Type: application/json" \
  -d '{"event_type": "test", "data": {}}'

References

• Dapr Pub/Sub
• Dapr Jobs API
• CloudEvents Spec

FastAPI Backend

Build production-grade FastAPI backends with SQLModel, Pydantic v2, and JWT/JWKS authentication patterns.

When to Use

• Building REST API endpoints with FastAPI
• Creating SQLModel schemas for Neon PostgreSQL
• Implementing JWT verification against Better Auth JWKS
• Designing OpenAPI contracts for frontend consumption
• Adding audit logging to API operations
• Ensuring human-agent parity in API design

Quick Start

# Project setup
uv init backend && cd backend
uv add fastapi sqlmodel pydantic httpx python-jose uvicorn

# Development
uv run uvicorn main:app --reload --port 8000

# Access docs
open http://localhost:8000/docs  # Swagger UI with Authorize button

Core Patterns

1. SQLModel Schema (Database + API)

SQLModel combines SQLAlchemy and Pydantic. Use table=True for database models:

from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional, Literal

# Base model (shared fields, no table)
class TaskBase(SQLModel):
    title: str = Field(max_length=200)
    description: Optional[str] = None
    status: Literal["pending", "in_progress", "review", "completed", "blocked"] = "pending"
    priority: Literal["low", "medium", "high", "critical"] = "medium"
    progress_percent: int = Field(default=0, ge=0, le=100)
    assigned_to: Optional[str] = None
    project_slug: Optional[str] = None
    parent_id: Optional[int] = None

# Database model (has table)
class Task(TaskBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)

# API models (no table, for request/response)
class TaskCreate(TaskBase):
    pass

class TaskUpdate(SQLModel):
    title: Optional[str] = None
    description: Optional[str] = None
    status: Optional[str] = None
    priority: Optional[str] = None
    progress_percent: Optional[int] = None
    assigned_to: Optional[str] = None

class TaskRead(TaskBase):
    id: int
    created_at: datetime
    updated_at: datetime

2. Neon PostgreSQL Connection

from sqlmodel import create_engine, Session
import os

# Neon connection string
DATABASE_URL = os.getenv("DATABASE_URL")  # postgresql://user:pass@host/db?sslmode=require

engine = create_engine(DATABASE_URL, echo=True)

def get_session():
    with Session(engine) as session:
        yield session

3. CRUD Endpoints

from fastapi import FastAPI, Depends, HTTPException, Query
from sqlmodel import Session, select

app = FastAPI(title="TaskFlow API", version="1.0.0")

@app.post("/api/tasks", response_model=TaskRead, status_code=201)
def create_task(
    task: TaskCreate,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
):
    db_task = Task.model_validate(task)
    session.add(db_task)
    session.commit()
    session.refresh(db_task)

    # Audit log
    log_action(session, "created", current_user.id, task_id=db_task.id)

    return db_task

@app.get("/api/tasks", response_model=list[TaskRead])
def list_tasks(
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
    status: Optional[str] = Query(None),
    assigned_to: Optional[str] = Query(None),
    project: Optional[str] = Query(None),
    limit: int = Query(50, le=100),
    offset: int = Query(0, ge=0),
):
    query = select(Task)

    if status:
        query = query.where(Task.status == status)
    if assigned_to:
        query = query.where(Task.assigned_to == assigned_to)
    if project:
        query = query.where(Task.project_slug == project)

    query = query.offset(offset).limit(limit)
    return session.exec(query).all()

@app.get("/api/tasks/{task_id}", response_model=TaskRead)
def get_task(
    task_id: int,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@app.patch("/api/tasks/{task_id}", response_model=TaskRead)
def update_task(
    task_id: int,
    task_update: TaskUpdate,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    update_data = task_update.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(task, key, value)

    task.updated_at = datetime.now()
    session.add(task)
    session.commit()
    session.refresh(task)

    # Audit log
    log_action(session, "updated", current_user.id, task_id=task.id, context=update_data)

    return task

4. JWT Authentication (Better Auth JWKS)

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, jwk, JWTError
import httpx
import time

security = HTTPBearer()

# Cache JWKS keys
_jwks_cache = None
_jwks_cache_time = 0
JWKS_CACHE_TTL = 3600  # 1 hour

AUTH_SERVER_URL = os.getenv("AUTH_SERVER_URL", "http://localhost:3001")
JWKS_URL = f"{AUTH_SERVER_URL}/api/auth/jwks"

async def get_jwks():
    """Fetch and cache JWKS from Better Auth server."""
    global _jwks_cache, _jwks_cache_time

    now = time.time()
    if _jwks_cache and (now - _jwks_cache_time) < JWKS_CACHE_TTL:
        return _jwks_cache

    async with httpx.AsyncClient() as client:
        response = await client.get(JWKS_URL)
        response.raise_for_status()
        _jwks_cache = response.json()
        _jwks_cache_time = now
        return _jwks_cache

async def verify_token(token: str) -> dict:
    """Verify JWT against Better Auth JWKS."""
    try:
        # Get JWKS
        jwks = await get_jwks()

        # Get unverified header to find key ID
        unverified_header = jwt.get_unverified_header(token)
        kid = unverified_header.get("kid")

        # Find matching key
        rsa_key = None
        for key in jwks.get("keys", []):
            if key.get("kid") == kid:
                rsa_key = key
                break

        if not rsa_key:
            raise HTTPException(status_code=401, detail="Key not found")

        # Verify token
        payload = jwt.decode(
            token,
            rsa_key,
            algorithms=["RS256"],
            options={"verify_aud": False}  # Adjust based on your setup
        )
        return payload

    except JWTError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"Invalid token: {str(e)}",
            headers={"WWW-Authenticate": "Bearer"},
        )

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    """Extract and verify current user from JWT."""
    token = credentials.credentials
    payload = await verify_token(token)

    return {
        "id": payload.get("sub"),
        "email": payload.get("email"),
        "role": payload.get("role", "user"),
    }

5. Swagger UI Authentication

FastAPI automatically adds an "Authorize" button when using HTTPBearer:

from fastapi import FastAPI
from fastapi.security import HTTPBearer

app = FastAPI(
    title="TaskFlow API",
    description="Human-Agent Task Management API",
    version="1.0.0",
)

# This adds the "Authorize" button to Swagger UI
security = HTTPBearer()

# Testing flow:
# 1. Login to your SSO (browser) → Get JWT
# 2. Open http://localhost:8000/docs
# 3. Click "Authorize" button
# 4. Paste JWT token (without "Bearer " prefix)
# 5. All requests now include Authorization header

6. Audit Logging

from sqlalchemy import JSON

class AuditLog(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    task_id: Optional[int] = None
    project_slug: Optional[str] = None
    actor_id: str
    actor_type: Literal["human", "agent"]
    action: str
    context: Optional[dict] = Field(default=None, sa_column_kwargs={"type_": JSON})
    timestamp: datetime = Field(default_factory=datetime.now)

def log_action(
    session: Session,
    action: str,
    actor_id: str,
    task_id: Optional[int] = None,
    project_slug: Optional[str] = None,
    context: Optional[dict] = None,
):
    """Create audit log entry."""
    # Determine actor type from worker registry
    worker = session.exec(select(Worker).where(Worker.id == actor_id)).first()
    actor_type = worker.type if worker else "human"

    log = AuditLog(
        task_id=task_id,
        project_slug=project_slug,
        actor_id=actor_id,
        actor_type=actor_type,
        action=action,
        context=context,
    )
    session.add(log)
    session.commit()
    return log

7. Agent Parity (MCP Compatibility)

Design endpoints that work for both CLI and MCP clients:

# Same endpoint serves:
# - CLI: taskflow start 1 → POST /api/tasks/1/start
# - MCP: claim_task(1) → POST /api/tasks/1/start
# - Web: Button click → POST /api/tasks/1/start

@app.post("/api/tasks/{task_id}/start", response_model=TaskRead)
def start_task(
    task_id: int,
    session: Session = Depends(get_session),
    current_user: dict = Depends(get_current_user),
):
    """Start a task (claim and begin work)."""
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    if task.status != "pending":
        raise HTTPException(
            status_code=400,
            detail=f"Cannot start task with status '{task.status}'"
        )

    task.status = "in_progress"
    task.assigned_to = current_user["id"]
    task.updated_at = datetime.now()

    session.add(task)
    session.commit()
    session.refresh(task)

    log_action(session, "started", current_user["id"], task_id=task.id)

    return task

@app.post("/api/tasks/{task_id}/progress", response_model=TaskRead)
def update_progress(
    task_id: int,
    percent: int = Query(..., ge=0, le=100),
    note: Optional[str] = Query(None),
    session: Session = Depends(get_session),
    current_user: dict = Depends(get_current_user),
):
    """Update task progress."""
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    if task.status != "in_progress":
        raise HTTPException(status_code=400, detail="Task must be in_progress")

    task.progress_percent = percent
    task.updated_at = datetime.now()

    session.add(task)
    session.commit()
    session.refresh(task)

    log_action(
        session, "progressed", current_user["id"],
        task_id=task.id,
        context={"percent": percent, "note": note}
    )

    return task

@app.post("/api/tasks/{task_id}/complete", response_model=TaskRead)
def complete_task(
    task_id: int,
    session: Session = Depends(get_session),
    current_user: dict = Depends(get_current_user),
):
    """Complete a task."""
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    if task.status not in ["in_progress", "review"]:
        raise HTTPException(
            status_code=400,
            detail=f"Cannot complete task with status '{task.status}'"
        )

    task.status = "completed"
    task.progress_percent = 100
    task.updated_at = datetime.now()

    session.add(task)
    session.commit()
    session.refresh(task)

    log_action(session, "completed", current_user["id"], task_id=task.id)

    return task

Project Structure

backend/
├── main.py              # FastAPI app, routes
├── models.py            # SQLModel schemas
├── database.py          # Neon connection
├── auth.py              # JWT/JWKS verification
├── audit.py             # Audit logging
├── dependencies.py      # Shared dependencies
└── tests/
    ├── conftest.py      # Test fixtures
    ├── test_tasks.py    # Task endpoint tests
    └── test_auth.py     # Auth tests

Critical: Async Session Patterns (MissingGreenlet Prevention)

After session.commit(), SQLAlchemy objects become detached. Accessing attributes triggers lazy loading which fails in async context with MissingGreenlet error.

The Pattern: Extract → Flush → Commit

@app.post("/api/tasks", response_model=TaskRead, status_code=201)
async def create_task(
    task: TaskCreate,
    session: AsyncSession = Depends(get_session),
    current_user: dict = Depends(get_current_user),
):
    # 1. Extract primitives from user BEFORE any commits
    actor_id = current_user["id"]

    # 2. Create entity and flush to get ID
    db_task = Task.model_validate(task)
    session.add(db_task)
    await session.flush()
    task_id = db_task.id  # Extract immediately after flush

    # 3. Call services with primitives (NOT objects)
    await log_action(session, actor_id=actor_id, task_id=task_id)

    # 4. Single commit at end
    await session.commit()
    await session.refresh(db_task)
    return db_task

Service Functions: Never Commit Internally

# WRONG - breaks caller's transaction
async def log_action(session: AsyncSession, ...):
    log = AuditLog(...)
    session.add(log)
    await session.commit()  # ❌ Caller loses control

# CORRECT - caller owns transaction
async def log_action(session: AsyncSession, ...):
    log = AuditLog(...)
    session.add(log)
    return log  # No commit

Input Validation for API Schemas

from pydantic import field_validator
from datetime import UTC, datetime

class TaskCreate(SQLModel):
    assignee_id: int | None = None
    due_date: datetime | None = None

    @field_validator("assignee_id", mode="after")
    @classmethod
    def zero_to_none(cls, v: int | None) -> int | None:
        """Swagger UI sends 0 for empty int fields."""
        return None if v == 0 else v

    @field_validator("due_date", mode="after")
    @classmethod
    def normalize_datetime(cls, v: datetime | None) -> datetime | None:
        """Strip timezone for naive UTC database columns."""
        if v and v.tzinfo:
            return v.astimezone(UTC).replace(tzinfo=None)
        return v

Testing with Pytest

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlmodel import SQLModel, create_engine, Session
from sqlmodel.pool import StaticPool

from main import app, get_session

@pytest.fixture
def session():
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

@pytest.fixture
def client(session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()

@pytest.fixture
def auth_headers():
    """Mock authenticated headers for testing."""
    return {"Authorization": "Bearer test-token"}

# tests/test_tasks.py
def test_create_task(client, auth_headers, mocker):
    # Mock auth
    mocker.patch("auth.get_current_user", return_value={"id": "@testuser"})

    response = client.post(
        "/api/tasks",
        json={"title": "Test Task"},
        headers=auth_headers,
    )
    assert response.status_code == 201
    assert response.json()["title"] == "Test Task"

Common Patterns

Error Handling

from fastapi import HTTPException
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.detail,
            "status_code": exc.status_code,
        },
    )

# Validation errors are handled automatically by Pydantic
# Returns 422 with detailed error messages

CORS Configuration

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=os.getenv("ALLOWED_ORIGINS", "http://localhost:3000").split(","),
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Health Check

@app.get("/health")
def health_check():
    return {"status": "healthy", "version": "1.0.0"}

@app.get("/api/health/db")
def db_health(session: Session = Depends(get_session)):
    try:
        session.exec(select(1))
        return {"database": "connected"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Database error: {e}")

References

For additional documentation, use Context7 MCP:

mcp__context7__resolve-library-id with libraryName="fastapi"
mcp__context7__get-library-docs with topic="authentication" or "sqlmodel"

See also: references/jwt-verification.md for detailed JWKS patterns.

Environment Variables

# Database
DATABASE_URL=postgresql://user:pass@host/db?sslmode=require

# Auth
AUTH_SERVER_URL=http://localhost:3001
JWKS_URL=http://localhost:3001/api/auth/jwks

# CORS
ALLOWED_ORIGINS=http://localhost:3000,http://localhost:3001

# Optional
LOG_LEVEL=INFO

Production Testing & Logging Patterns

Structured Logging

Setup with structlog

# app/logging.py
import structlog
import logging
from contextvars import ContextVar

request_id_ctx: ContextVar[str] = ContextVar("request_id", default="")

def configure_logging():
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer()
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        context_class=dict,
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=True,
    )

def get_logger():
    return structlog.get_logger()

Request ID Middleware

# app/middleware.py
import uuid
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from app.logging import request_id_ctx, get_logger

class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request_id_ctx.set(request_id)

        log = get_logger()
        log.info(
            "request_started",
            method=request.method,
            path=request.url.path,
            request_id=request_id
        )

        response = await call_next(request)

        log.info(
            "request_completed",
            method=request.method,
            path=request.url.path,
            status_code=response.status_code,
            request_id=request_id
        )

        response.headers["X-Request-ID"] = request_id
        return response

Usage in Routes

from app.logging import get_logger

@router.post("/tasks")
async def create_task(task: TaskCreate, session: AsyncSession = Depends(get_session)):
    log = get_logger()
    log.info("creating_task", title=task.title)

    db_task = Task.model_validate(task)
    session.add(db_task)
    await session.commit()
    await session.refresh(db_task)

    log.info("task_created", task_id=db_task.id)
    return db_task

Testing Patterns

conftest.py Setup

# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlmodel import SQLModel

from app.main import app
from app.database import get_session

# Use test database
TEST_DATABASE_URL = "postgresql+asyncpg://test:test@localhost:5432/test_db"

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

@pytest.fixture(scope="session")
async def engine():
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)
    await engine.dispose()

@pytest.fixture
async def session(engine):
    async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    async with async_session() as session:
        yield session
        await session.rollback()

@pytest.fixture
async def client(session):
    def override_get_session():
        yield session

    app.dependency_overrides[get_session] = override_get_session

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

    app.dependency_overrides.clear()

Test Examples

# tests/test_tasks.py
import pytest
from httpx import AsyncClient

@pytest.mark.anyio
async def test_create_task(client: AsyncClient):
    response = await client.post("/tasks", json={
        "title": "Test Task",
        "status": "pending"
    })
    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Test Task"
    assert "id" in data

@pytest.mark.anyio
async def test_get_task_not_found(client: AsyncClient):
    response = await client.get("/tasks/99999")
    assert response.status_code == 404

@pytest.mark.anyio
async def test_list_tasks_pagination(client: AsyncClient):
    # Create 15 tasks
    for i in range(15):
        await client.post("/tasks", json={"title": f"Task {i}"})

    # Test pagination
    response = await client.get("/tasks?limit=10&offset=0")
    assert response.status_code == 200
    data = response.json()
    assert len(data["items"]) == 10
    assert data["total"] >= 15

Testing with Mocked Dependencies

# tests/test_with_mocks.py
from unittest.mock import AsyncMock, patch
import pytest

@pytest.mark.anyio
async def test_external_api_call(client: AsyncClient):
    mock_response = {"status": "success", "data": {"id": 123}}

    with patch("app.services.external_api.fetch_data", new_callable=AsyncMock) as mock_fetch:
        mock_fetch.return_value = mock_response

        response = await client.post("/sync-external")

        assert response.status_code == 200
        mock_fetch.assert_called_once()

Dependency Injection Patterns

Repository Pattern

# app/repositories/task.py
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from app.models import Task, TaskCreate, TaskUpdate

class TaskRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def create(self, task: TaskCreate) -> Task:
        db_task = Task.model_validate(task)
        self.session.add(db_task)
        await self.session.commit()
        await self.session.refresh(db_task)
        return db_task

    async def get(self, task_id: int) -> Task | None:
        return await self.session.get(Task, task_id)

    async def list(self, limit: int = 10, offset: int = 0) -> list[Task]:
        result = await self.session.exec(
            select(Task).offset(offset).limit(limit)
        )
        return result.all()

Service Layer

# app/services/task.py
from app.repositories.task import TaskRepository
from app.models import Task, TaskCreate
from app.logging import get_logger

class TaskService:
    def __init__(self, repo: TaskRepository):
        self.repo = repo
        self.log = get_logger()

    async def create_task(self, task: TaskCreate, user_id: str) -> Task:
        self.log.info("creating_task", user_id=user_id, title=task.title)
        db_task = await self.repo.create(task)
        # Additional business logic here
        return db_task

Dependency Factory

# app/dependencies.py
from fastapi import Depends
from sqlmodel.ext.asyncio.session import AsyncSession
from app.database import get_session
from app.repositories.task import TaskRepository
from app.services.task import TaskService

def get_task_repository(session: AsyncSession = Depends(get_session)) -> TaskRepository:
    return TaskRepository(session)

def get_task_service(repo: TaskRepository = Depends(get_task_repository)) -> TaskService:
    return TaskService(repo)

Usage in Routes

# app/routers/tasks.py
from fastapi import APIRouter, Depends
from app.dependencies import get_task_service
from app.services.task import TaskService

router = APIRouter(prefix="/tasks", tags=["tasks"])

@router.post("/", response_model=TaskRead, status_code=201)
async def create_task(
    task: TaskCreate,
    service: TaskService = Depends(get_task_service),
    user: dict = Depends(get_current_user)
):
    return await service.create_task(task, user["sub"])

API Versioning

URL Prefix Versioning

# app/main.py
from fastapi import FastAPI
from app.routers.v1 import tasks as tasks_v1
from app.routers.v2 import tasks as tasks_v2

app = FastAPI()

app.include_router(tasks_v1.router, prefix="/v1")
app.include_router(tasks_v2.router, prefix="/v2")

Header-Based Versioning

from fastapi import Header, HTTPException

async def get_api_version(x_api_version: str = Header(default="1")):
    if x_api_version not in ["1", "2"]:
        raise HTTPException(status_code=400, detail="Invalid API version")
    return x_api_version

@router.get("/tasks")
async def list_tasks(version: str = Depends(get_api_version)):
    if version == "2":
        return {"items": tasks, "total": len(tasks), "version": 2}
    return tasks  # v1 format

Error Handling

Custom Exception Handlers

# app/exceptions.py
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
from app.logging import get_logger

class AppException(Exception):
    def __init__(self, status_code: int, detail: str, error_code: str):
        self.status_code = status_code
        self.detail = detail
        self.error_code = error_code

async def app_exception_handler(request: Request, exc: AppException):
    log = get_logger()
    log.error(
        "app_exception",
        error_code=exc.error_code,
        detail=exc.detail,
        path=request.url.path
    )
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.error_code,
            "detail": exc.detail
        }
    )

# In main.py
app.add_exception_handler(AppException, app_exception_handler)

Production Checklist

Logging

• [ ] Structured JSON logging configured
• [ ] Request ID middleware added
• [ ] Correlation IDs propagated to downstream services
• [ ] Sensitive data not logged (passwords, tokens)

Testing

• [ ] pytest with anyio for async tests
• [ ] Test database with transaction rollback
• [ ] Dependency override for mocking
• [ ] Integration tests with real DB
• [ ] Coverage > 80%

Architecture

• [ ] Repository pattern for data access
• [ ] Service layer for business logic
• [ ] Dependency injection with Depends()
• [ ] API versioning strategy defined

Error Handling

• [ ] Custom exception classes
• [ ] Global exception handlers
• [ ] Consistent error response format
• [ ] Errors logged with context

SQLModel Database Patterns

Design and implement database schemas using SQLModel - combining SQLAlchemy's power with Pydantic's validation.

Quick Start

# Async support (recommended for FastAPI)
uv add sqlmodel sqlalchemy[asyncio] asyncpg  # PostgreSQL
# or
uv add sqlmodel sqlalchemy[asyncio] aiosqlite  # SQLite

---

Core Patterns

1. Model Hierarchy (Base → Table → API)

from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional, Literal

# Base model - shared fields, validation, NO table
class TaskBase(SQLModel):
    title: str = Field(max_length=200, index=True)
    description: Optional[str] = None
    status: Literal["pending", "in_progress", "completed"] = "pending"
    priority: Literal["low", "medium", "high"] = "medium"

# Table model - has table=True, adds id and timestamps
class Task(TaskBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)

# Create model - for POST requests
class TaskCreate(TaskBase):
    pass

# Update model - all fields optional for PATCH
class TaskUpdate(SQLModel):
    title: Optional[str] = None
    description: Optional[str] = None
    status: Optional[str] = None

# Read model - for responses
class TaskRead(TaskBase):
    id: int
    created_at: datetime

2. Async Database Connection

from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
import os

# postgresql+asyncpg://user:pass@host/db
DATABASE_URL = os.getenv("DATABASE_URL")

async_engine = create_async_engine(DATABASE_URL, echo=True)

async def create_db_and_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

async def get_session() -> AsyncSession:
    async with AsyncSession(async_engine) as session:
        yield session

3. Relationships

One-to-Many

from sqlmodel import Relationship

class Project(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    tasks: List["Task"] = Relationship(back_populates="project")

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    project_id: Optional[int] = Field(default=None, foreign_key="project.id")
    project: Optional[Project] = Relationship(back_populates="tasks")

Many-to-Many

class TaskWorkerLink(SQLModel, table=True):
    task_id: Optional[int] = Field(default=None, foreign_key="task.id", primary_key=True)
    worker_id: Optional[str] = Field(default=None, foreign_key="worker.id", primary_key=True)

class Worker(SQLModel, table=True):
    id: str = Field(primary_key=True)
    name: str
    tasks: List["Task"] = Relationship(back_populates="workers", link_model=TaskWorkerLink)

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    workers: List[Worker] = Relationship(back_populates="tasks", link_model=TaskWorkerLink)

4. Query Patterns

from sqlmodel import select, or_

# Basic queries
statement = select(Task).where(Task.status == "pending")
results = await session.exec(statement)
tasks = results.all()

# Multiple conditions
statement = select(Task).where(Task.status == "pending", Task.priority == "high")

# OR conditions
statement = select(Task).where(or_(Task.status == "pending", Task.status == "in_progress"))

# Ordering and pagination
statement = select(Task).order_by(Task.created_at.desc()).offset(20).limit(10)

# Get by ID
task = await session.get(Task, task_id)

# Eager loading relationships
from sqlalchemy.orm import selectinload
statement = select(Task).options(selectinload(Task.project))

5. Neon PostgreSQL Connection

DATABASE_URL = os.getenv("DATABASE_URL")

# Convert sync URL to async
if DATABASE_URL.startswith("postgresql://"):
    DATABASE_URL = DATABASE_URL.replace(
        "postgresql://", "postgresql+asyncpg://", 1
    ).replace("sslmode=", "ssl=")

async_engine = create_async_engine(DATABASE_URL, pool_size=5, max_overflow=10)

---

Migrations with Alembic

Setup

uv add alembic
alembic init migrations

Configuration

Edit migrations/env.py:

from sqlmodel import SQLModel
from app.models import *  # Import all models

target_metadata = SQLModel.metadata

Commands

# Create migration
alembic revision --autogenerate -m "Add tasks table"

# Apply migrations
alembic upgrade head

# Rollback one step
alembic downgrade -1

Common Operations

# Add column
op.add_column('task', sa.Column('priority', sa.String(), default='medium'))

# Add index
op.create_index('ix_task_status', 'task', ['status'])

# Add foreign key
op.add_column('task', sa.Column('project_id', sa.Integer()))
op.create_foreign_key('fk_task_project', 'task', 'project', ['project_id'], ['id'])

---

Critical Patterns

Always use `await session.exec()` not `session.execute()` for SQLModel select statements.

# Correct
results = await session.exec(select(Task))

# Also correct for getting by ID
task = await session.get(Task, task_id)

Testing Pattern

import pytest
from sqlmodel import create_engine, Session
from sqlmodel.pool import StaticPool

@pytest.fixture
def session():
    """In-memory SQLite for fast tests."""
    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session