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
python
1# execution/schemas/invoice.py
2from pydantic import BaseModel, EmailStr, Field
3from typing import Optional
4from datetime import datetime
5
6class LineItem(BaseModel):
7 description: str
8 quantity: int = Field(gt=0)
9 unit_price: float = Field(gt=0)
10
11class CreateInvoiceRequest(BaseModel):
12 customer_email: EmailStr
13 customer_name: str = Field(min_length=1, max_length=100)
14 line_items: list[LineItem] = Field(min_length=1)
15 due_date: datetime
16 memo: Optional[str] = None
17
18class InvoiceResponse(BaseModel):
19 invoice_id: str
20 qbo_id: Optional[str] = None
21 status: str
22 total: float
23 created_at: datetime
2. Create the Endpoint
python
1# execution/api/invoices.py
2from fastapi import APIRouter, HTTPException, Depends
3from ..schemas.invoice import CreateInvoiceRequest, InvoiceResponse
4from ..services.qbo_service import qbo_service
5import logging
6
7router = APIRouter(prefix="/invoices", tags=["invoices"])
8log = logging.getLogger(__name__)
9
10@router.post("/", response_model=InvoiceResponse, status_code=201)
11async def create_invoice(request: CreateInvoiceRequest):
12 """Create a new invoice in QuickBooks."""
13 try:
14 # Calculate total
15 total = sum(item.quantity * item.unit_price for item in request.line_items)
16
17 # Create in QBO
18 qbo_invoice = await qbo_service.create_invoice(
19 customer_email=request.customer_email,
20 customer_name=request.customer_name,
21 line_items=[item.model_dump() for item in request.line_items],
22 due_date=request.due_date,
23 memo=request.memo
24 )
25
26 log.info("Invoice created", extra={
27 "invoice_id": qbo_invoice.id,
28 "customer": request.customer_email,
29 "total": total
30 })
31
32 return InvoiceResponse(
33 invoice_id=qbo_invoice.doc_number,
34 qbo_id=qbo_invoice.id,
35 status="created",
36 total=total,
37 created_at=datetime.utcnow()
38 )
39
40 except QBOAuthError:
41 raise HTTPException(status_code=401, detail="QuickBooks auth expired")
42 except QBOValidationError as e:
43 raise HTTPException(status_code=422, detail=str(e))
44 except Exception as e:
45 log.exception("Invoice creation failed")
46 raise HTTPException(status_code=500, detail="Internal error")
3. Wire Up the Router
python
1# execution/main.py
2from fastapi import FastAPI
3from .api import invoices, webhooks, inventory
4
5app = FastAPI(
6 title="SMB Automation API",
7 version="1.0.0"
8)
9
10app.include_router(invoices.router, prefix="/api/v1")
11app.include_router(webhooks.router, prefix="/api/v1")
12app.include_router(inventory.router, prefix="/api/v1")
13
14@app.get("/health")
15async def health():
16 return {"status": "ok"}
Common Endpoint Patterns
Webhook Receiver
python
1# execution/api/webhooks.py
2from fastapi import APIRouter, Request, HTTPException, Header
3import hmac
4import hashlib
5
6router = APIRouter(prefix="/webhooks", tags=["webhooks"])
7
8@router.post("/stripe")
9async def stripe_webhook(
10 request: Request,
11 stripe_signature: str = Header(alias="Stripe-Signature")
12):
13 """Handle Stripe webhook events."""
14 payload = await request.body()
15
16 # Verify signature
17 try:
18 event = stripe.Webhook.construct_event(
19 payload, stripe_signature, WEBHOOK_SECRET
20 )
21 except stripe.error.SignatureVerificationError:
22 raise HTTPException(status_code=400, detail="Invalid signature")
23
24 # Process idempotently
25 event_id = event["id"]
26 if await is_processed(event_id):
27 return {"status": "duplicate"}
28
29 # Handle event types
30 match event["type"]:
31 case "payment_intent.succeeded":
32 await handle_payment_success(event["data"]["object"])
33 case "invoice.paid":
34 await handle_invoice_paid(event["data"]["object"])
35
36 await mark_processed(event_id)
37 return {"status": "processed"}
CRUD Resource
python
1# execution/api/customers.py
2from fastapi import APIRouter, HTTPException, Query
3from typing import Optional
4
5router = APIRouter(prefix="/customers", tags=["customers"])
6
7@router.get("/")
8async def list_customers(
9 skip: int = Query(0, ge=0),
10 limit: int = Query(50, ge=1, le=100),
11 status: Optional[str] = None
12):
13 """List customers with pagination."""
14 customers = await customer_service.list(skip=skip, limit=limit, status=status)
15 return {"customers": customers, "total": len(customers)}
16
17@router.get("/{customer_id}")
18async def get_customer(customer_id: str):
19 """Get customer by ID."""
20 customer = await customer_service.get(customer_id)
21 if not customer:
22 raise HTTPException(status_code=404, detail="Customer not found")
23 return customer
24
25@router.post("/", status_code=201)
26async def create_customer(request: CreateCustomerRequest):
27 """Create new customer."""
28 customer = await customer_service.create(request)
29 return customer
30
31@router.patch("/{customer_id}")
32async def update_customer(customer_id: str, request: UpdateCustomerRequest):
33 """Update customer."""
34 customer = await customer_service.update(customer_id, request)
35 if not customer:
36 raise HTTPException(status_code=404, detail="Customer not found")
37 return customer
38
39@router.delete("/{customer_id}", status_code=204)
40async def delete_customer(customer_id: str):
41 """Delete customer."""
42 deleted = await customer_service.delete(customer_id)
43 if not deleted:
44 raise HTTPException(status_code=404, detail="Customer not found")
Background Task Trigger
python
1# execution/api/jobs.py
2from fastapi import APIRouter, BackgroundTasks
3
4router = APIRouter(prefix="/jobs", tags=["jobs"])
5
6@router.post("/sync-inventory")
7async def trigger_inventory_sync(background_tasks: BackgroundTasks):
8 """Trigger background inventory sync."""
9 job_id = generate_job_id()
10
11 background_tasks.add_task(sync_inventory_task, job_id)
12
13 return {
14 "job_id": job_id,
15 "status": "queued",
16 "check_status": f"/api/v1/jobs/{job_id}"
17 }
18
19@router.get("/{job_id}")
20async def get_job_status(job_id: str):
21 """Check job status."""
22 job = await job_store.get(job_id)
23 if not job:
24 raise HTTPException(status_code=404, detail="Job not found")
25 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
txt
1# requirements.txt
2fastapi>=0.109.0
3uvicorn>=0.27.0
4pydantic>=2.5.0
5pydantic-settings>=2.1.0
6httpx>=0.26.0
7python-dotenv>=1.0.0
Running Locally
bash
1# Install
2pip install -r requirements.txt
3
4# Run
5uvicorn execution.main:app --reload --port 8000
6
7# Docs
8open http://localhost:8000/docs
Deployment Checklist
