API Scaffolding for SMB Automation

API Scaffolding for SMB Automation

<!-- SKILL METADATA skill_id: api-scaffolding version: 2.0.0 domain: development reviewed: 2026-03-06 status: READY changelog: 2.0.0 - Broadened description, added pre-flight checklist and decision tree, removed redundant When-to-Use 1.0.0 - Initial version stamp (auto-generated by stamp_skill_headers.py) -->

Spin up professional FastAPI endpoints in minutes. Start with the pre-flight checklist — knowing these answers before writing code prevents the most common rework.

Pre-Flight Checklist

Before scaffolding anything, confirm:

• Does a script already exist? Check execution/ — you may be wrapping an existing script, not starting from scratch
• What are the inputs? (form fields, webhook payload, query params)
• What does success return? (ID, status, redirect, file?)
• What errors need to be surfaced? (validation, external API failure, not-found)
• Is this public or internal? Public endpoints need rate limiting; internal can trust the caller
• Does this trigger an external API? (QBO, ShipStation, Stripe — if yes, read the relevant integration skill)

Decision Tree: What Type of Endpoint?

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

Before handing off to client:

• All endpoints have Pydantic request/​response models
• Error handling returns proper HTTP status codes (400 validation, 401 auth, 404 not found, 422 upstream, 503 dependency down)
• Logging includes context (request ID, user/​client, key IDs)
• Secrets are in .env, never hardcoded
• Health endpoint at /​health returns {"status": "ok"}
• CORS configured if called from a browser
• Rate limiting on any public-facing endpoints
• Tested locally with --reload and confirmed /​docs renders
• .env.example documents every required key