Why integration is the job

An AI system is only valuable when it's connected — to the CRM, the helpdesk, the data warehouse, the website. Track 9 is the "forward-deployed" flavor of the market: getting AI into the systems a business already runs. The two fundamental integration patterns are REST (you ask another system, or it asks you, synchronously) and webhooks (a system tells you something happened, asynchronously). Most AI integrations — including the hybrid code-service from Tracks 4 and 8 — are built from these two.

REST: request/response

REST is synchronous request/response over HTTP. You expose your AI capability as an endpoint others call, and you call others' endpoints. The pieces you must get right:

• Verbs — GET (read, no side effects), POST (create/act), PUT/PATCH (update), DELETE. An AI action endpoint is almost always POST.
• Status codes — 200 OK, 201 created, 400 bad request, 401 unauthorized, 429 rate-limited, 500 server error. Return the right code so callers can react correctly (retry a 429, don't retry a 400).
• Auth — never expose an unauthenticated AI endpoint (it costs you tokens per call). Common: an API key or bearer token in a header; OAuth for user-delegated access.

from fastapi import FastAPI, Header, HTTPException
app = FastAPI()

@app.post("/classify")                                   # POST: it performs an action (and costs tokens)
def classify(payload: dict, authorization: str = Header(None)):
    if authorization != f"Bearer {EXPECTED_TOKEN}":      # AUTH first — reject before spending a token
        raise HTTPException(status_code=401, detail="unauthorized")   # 401, the correct code
    if "text" not in payload:                            # validate input
        raise HTTPException(status_code=400, detail="missing 'text'") # 400 = caller's mistake, don't retry
    return {"category": my_classifier(payload["text"])}  # 200 with the result

This is the same code-service you exposed in the hybrid pattern (Track 8) — now with the auth and status-code rigor a real caller depends on.

Webhooks: the async inverse

A REST call is you asking. A webhook is another system telling you — it POSTs to your endpoint when an event happens (a new support ticket, a form submission, a payment). Webhooks are how AI systems react to the world in real time without polling: "a ticket arrived → classify and route it." (Every N8N/Zapier trigger is a webhook under the hood.)

@app.post("/webhooks/new-ticket")                        # a URL you register with the source system
def on_new_ticket(event: dict):
    # The source system calls THIS when a ticket is created — you react to the event.
    ticket = event["data"]
    category = my_classifier(ticket["body"])             # run your AI on the incoming event
    route_ticket(ticket["id"], category)                 # act on the result
    return {"received": True}                             # ack quickly (see: respond fast, work async)

Verify the webhook is real

Your webhook URL is public — anyone who finds it can POST fake events (spoofed tickets, forged payments). So you must verify each webhook actually came from the expected sender, usually via a signature: the sender signs the payload with a shared secret, and you recompute and compare:

import hmac, hashlib

def verify(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()  # recompute the HMAC of the raw body
    return hmac.compare_digest(expected, signature)      # constant-time compare (avoids timing attacks)

@app.post("/webhooks/new-ticket")
async def on_new_ticket(request: Request):
    body = await request.body()                          # the RAW bytes — verify BEFORE parsing
    sig = request.headers.get("X-Signature", "")
    if not verify(body, sig, WEBHOOK_SECRET):            # reject anything not signed by the real sender
        raise HTTPException(status_code=401, detail="bad signature")
    event = json.loads(body)
    # … process the now-trusted event …

Unverified webhook input is also untrusted input to your AI — it can carry an indirect prompt injection (Track 6). Verify the sender and treat the content as untrusted.

Idempotency: survive retries

Networks are unreliable, so senders retry — and REST/webhook calls will be delivered more than once. If processing a duplicate causes a second charge, a second email, or a second DB row, that's a real bug. Idempotency means: processing the same request twice has the same effect as processing it once. The standard mechanism is an idempotency key — a unique id per logical event that you record and check:

processed = set()                                        # a store of handled event ids (a DB table in production)

@app.post("/webhooks/payment")
def on_payment(event: dict):
    event_id = event["id"]                               # the sender's unique id for THIS event
    if event_id in processed:                            # already handled? (a retry / duplicate delivery)
        return {"status": "duplicate, ignored"}          # no-op — safe to receive twice
    processed.add(event_id)                              # record it BEFORE the side effect
    send_confirmation_email(event["customer"])           # the side effect runs exactly once per event_id
    return {"status": "ok"}

Design any endpoint with side effects to be idempotent: dedupe on the sender's event id (or require an Idempotency-Key header on your own APIs, as Stripe does). This is the single most important reliability property for integrations — and the same discipline appears in the N8N pipelines you may already run.

Sync REST vs. webhook — choosing

• REST (synchronous) when the caller needs the answer now to continue — a user waiting on a classification, a chat response. The caller blocks on your response.
• Webhook (asynchronous / event-driven) when something happened and you react without the source waiting — a new ticket, a nightly export, a status change. Decouples the systems.

Many real integrations use both: a webhook notifies you of an event, you ack fast, process async, and maybe call back via REST (or another webhook) with the result. Matching the pattern to whether the caller is waiting is the core design decision.

Pitfalls

• Unauthenticated AI endpoints. A public, keyless endpoint burns your tokens (and budget) for anyone who finds it — always auth.
• Wrong status codes. Returning 200 on failure (or 500 on a caller's bad input) breaks callers' retry logic — return the right code.
• Trusting unverified webhooks. Public URLs get spoofed — verify signatures, and treat content as untrusted AI input (injection).
• Slow inline webhook handlers. Doing the LLM call before acking causes sender timeouts and retries — ack fast, work async.
• Non-idempotent side effects. Retries then double-charge/double-send — dedupe on an idempotency key.

Recap

• AI value comes from integration; the two primitives are REST (synchronous request/response) and webhooks (async, a system notifying you of events).
• Get REST right: correct verbs and status codes, and never expose an unauthenticated AI endpoint.
• Verify webhook signatures (HMAC) before trusting a payload — and treat the content as untrusted AI input.
• Ack webhooks fast, process async, and make side-effecting handlers idempotent (dedupe on an event/ idempotency key) so retries are harmless.
• Choose REST when the caller waits, webhooks when reacting to events; real integrations often combine both.
• Worked examples map to the exercises: an authed REST endpoint (Ex 1), signature verification (Ex 2), idempotency (Ex 3), REST-vs-webhook choice (Ex 4).

from fastapi import FastAPI, Header, HTTPException
app = FastAPI()
EXPECTED = "secret-token"

@app.post("/classify")
def classify(payload: dict, authorization: str = Header(None)):
    if authorization != f"Bearer {EXPECTED}":
        raise HTTPException(401, "unauthorized")     # 401 for missing/wrong auth — reject before spending a token
    if "text" not in payload:
        raise HTTPException(400, "missing 'text'")   # 400 = caller's mistake (don't retry)
    return {"category": my_classifier(payload["text"])}  # 200 with result

import hmac, hashlib, json
from fastapi import Request, HTTPException
SECRET = b"webhook-secret"

def verify(body: bytes, sig: str) -> bool:
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, sig)   # constant-time compare

@app.post("/webhooks/ticket")
async def on_ticket(request: Request):
    body = await request.body()                 # RAW bytes — verify before parsing
    if not verify(body, request.headers.get("X-Signature", "")):
        raise HTTPException(401, "bad signature")
    event = json.loads(body)
    return {"received": True}

processed = set()    # a DB table in production

@app.post("/webhooks/payment")
def on_payment(event: dict):
    eid = event["id"]
    if eid in processed:
        return {"status": "duplicate, ignored"}   # retry / double-delivery -> no-op
    processed.add(eid)                             # record BEFORE the side effect
    send_confirmation_email(event["customer"])     # runs exactly once per event id
    return {"status": "ok"}

# Replay proof:
on_payment({"id": "evt_1", "customer": "a@b.com"})  # sends email
on_payment({"id": "evt_1", "customer": "a@b.com"})  # ignored — no second email