name: api-scaffolding description: Quickly scaffold production-ready FastAPI endpoints for client workflows. Use when building new APIs for QuickBooks, ShipStation, or webhook integrations.

API Scaffolding for SMB Automation

Spin up professional REST APIs in minutes for client deliverables.

When to Use This Skill

• Creating new API endpoints for client workflows
• Building webhook receivers (Stripe, ShipStation, etc.)
• Setting up FastAPI projects from scratch
• Adding endpoints to existing execution/ scripts

Quick Start: New Endpoint

1. Define the Pydantic Models

# execution/schemas/invoice.py
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime

class LineItem(BaseModel):
    description: str
    quantity: int = Field(gt=0)
    unit_price: float = Field(gt=0)

class CreateInvoiceRequest(BaseModel):
    customer_email: EmailStr
    customer_name: str = Field(min_length=1, max_length=100)
    line_items: list[LineItem] = Field(min_length=1)
    due_date: datetime
    memo: Optional[str] = None

class InvoiceResponse(BaseModel):
    invoice_id: str
    qbo_id: Optional[str] = None
    status: str
    total: float
    created_at: datetime

2. Create the Endpoint

# execution/api/invoices.py
from fastapi import APIRouter, HTTPException, Depends
from ..schemas.invoice import CreateInvoiceRequest, InvoiceResponse
from ..services.qbo_service import qbo_service
import logging

router = APIRouter(prefix="/invoices", tags=["invoices"])
log = logging.getLogger(__name__)

@router.post("/", response_model=InvoiceResponse, status_code=201)
async def create_invoice(request: CreateInvoiceRequest):
    """Create a new invoice in QuickBooks."""
    try:
        # Calculate total
        total = sum(item.quantity * item.unit_price for item in request.line_items)

        # Create in QBO
        qbo_invoice = await qbo_service.create_invoice(
            customer_email=request.customer_email,
            customer_name=request.customer_name,
            line_items=[item.model_dump() for item in request.line_items],
            due_date=request.due_date,
            memo=request.memo
        )

        log.info("Invoice created", extra={
            "invoice_id": qbo_invoice.id,
            "customer": request.customer_email,
            "total": total
        })

        return InvoiceResponse(
            invoice_id=qbo_invoice.doc_number,
            qbo_id=qbo_invoice.id,
            status="created",
            total=total,
            created_at=datetime.utcnow()
        )

    except QBOAuthError:
        raise HTTPException(status_code=401, detail="QuickBooks auth expired")
    except QBOValidationError as e:
        raise HTTPException(status_code=422, detail=str(e))
    except Exception as e:
        log.exception("Invoice creation failed")
        raise HTTPException(status_code=500, detail="Internal error")

3. Wire Up the Router

# execution/main.py
from fastapi import FastAPI
from .api import invoices, webhooks, inventory

app = FastAPI(
    title="SMB Automation API",
    version="1.0.0"
)

app.include_router(invoices.router, prefix="/api/v1")
app.include_router(webhooks.router, prefix="/api/v1")
app.include_router(inventory.router, prefix="/api/v1")

@app.get("/health")
async def health():
    return {"status": "ok"}

Common Endpoint Patterns

Webhook Receiver

# execution/api/webhooks.py
from fastapi import APIRouter, Request, HTTPException, Header
import hmac
import hashlib

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

@router.post("/stripe")
async def stripe_webhook(
    request: Request,
    stripe_signature: str = Header(alias="Stripe-Signature")
):
    """Handle Stripe webhook events."""
    payload = await request.body()

    # Verify signature
    try:
        event = stripe.Webhook.construct_event(
            payload, stripe_signature, WEBHOOK_SECRET
        )
    except stripe.error.SignatureVerificationError:
        raise HTTPException(status_code=400, detail="Invalid signature")

    # Process idempotently
    event_id = event["id"]
    if await is_processed(event_id):
        return {"status": "duplicate"}

    # Handle event types
    match event["type"]:
        case "payment_intent.succeeded":
            await handle_payment_success(event["data"]["object"])
        case "invoice.paid":
            await handle_invoice_paid(event["data"]["object"])

    await mark_processed(event_id)
    return {"status": "processed"}

CRUD Resource

# execution/api/customers.py
from fastapi import APIRouter, HTTPException, Query
from typing import Optional

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

@router.get("/")
async def list_customers(
    skip: int = Query(0, ge=0),
    limit: int = Query(50, ge=1, le=100),
    status: Optional[str] = None
):
    """List customers with pagination."""
    customers = await customer_service.list(skip=skip, limit=limit, status=status)
    return {"customers": customers, "total": len(customers)}

@router.get("/{customer_id}")
async def get_customer(customer_id: str):
    """Get customer by ID."""
    customer = await customer_service.get(customer_id)
    if not customer:
        raise HTTPException(status_code=404, detail="Customer not found")
    return customer

@router.post("/", status_code=201)
async def create_customer(request: CreateCustomerRequest):
    """Create new customer."""
    customer = await customer_service.create(request)
    return customer

@router.patch("/{customer_id}")
async def update_customer(customer_id: str, request: UpdateCustomerRequest):
    """Update customer."""
    customer = await customer_service.update(customer_id, request)
    if not customer:
        raise HTTPException(status_code=404, detail="Customer not found")
    return customer

@router.delete("/{customer_id}", status_code=204)
async def delete_customer(customer_id: str):
    """Delete customer."""
    deleted = await customer_service.delete(customer_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Customer not found")

Background Task Trigger

# execution/api/jobs.py
from fastapi import APIRouter, BackgroundTasks

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

@router.post("/sync-inventory")
async def trigger_inventory_sync(background_tasks: BackgroundTasks):
    """Trigger background inventory sync."""
    job_id = generate_job_id()

    background_tasks.add_task(sync_inventory_task, job_id)

    return {
        "job_id": job_id,
        "status": "queued",
        "check_status": f"/api/v1/jobs/{job_id}"
    }

@router.get("/{job_id}")
async def get_job_status(job_id: str):
    """Check job status."""
    job = await job_store.get(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found")
    return job

Project Structure

execution/
├── main.py                 # FastAPI app entry
├── api/
│   ├── __init__.py
│   ├── invoices.py         # Invoice endpoints
│   ├── webhooks.py         # Webhook receivers
│   ├── customers.py        # Customer CRUD
│   └── jobs.py             # Background job triggers
├── schemas/
│   ├── __init__.py
│   ├── invoice.py          # Invoice Pydantic models
│   └── customer.py         # Customer Pydantic models
├── services/
│   ├── __init__.py
│   ├── qbo_service.py      # QuickBooks API client
│   └── stripe_service.py   # Stripe API client
└── core/
    ├── config.py           # Settings from .env
    ├── database.py         # DB connection (if needed)
    └── security.py         # Auth helpers

Essential Dependencies

# requirements.txt
fastapi>=0.109.0
uvicorn>=0.27.0
pydantic>=2.5.0
pydantic-settings>=2.1.0
httpx>=0.26.0
python-dotenv>=1.0.0

Running Locally

# Install
pip install -r requirements.txt

# Run
uvicorn execution.main:app --reload --port 8000

# Docs
open http://localhost:8000/docs

Deployment Checklist

• All endpoints have Pydantic request/response models
• Error handling returns proper status codes
• Logging includes context (IDs, user info)
• Secrets in .env, not hardcoded
• Health endpoint at /health
• CORS configured if needed
• Rate limiting for public endpoints