Lesson 16.29: Streaming, Webhooks & Real-Time AI Workflows

01 Deconstructing the Problem: The Danger of Silent Waiting in AI

Over the previous two lessons, we established a resilient backend foundation: clean layered architecture (Lesson 27) and Celery + Redis background queuing with retry engineering (Lesson 28). However, from the user experience perspective, a critical architectural void remains.

Consider an enterprise analyst in Byteion SaaS uploading a 200-page financial statement and asking:

“Extract all hidden financial risk vectors and statutory compliance violations across this document.”

Fulfilling this query requires eight discrete computational stages: PDF extraction, OCR fallback, chunking, embedding generation, Qdrant vector retrieval, multi-turn agent reasoning, validation, and final report compilation. This workload requires between 45 seconds and two minutes.

We must deliver two distinct real-time feedback mechanisms:

02 Comparative Evaluation: Polling, Streaming, WebSockets vs. Webhooks

Selecting the correct communication primitive is vital to preventing architectural bloat and resource exhaustion:

03 Enterprise Real-Time AI Architecture Blueprint

The end-to-end telemetry and event-driven architecture is structured across decoupled pipelines:

Three Non-Negotiable Data Retention Rules:

1. PostgreSQL Job Table: Authoritative master for final business states and durable results.
2. Event Stream (Redis Streams / Event Table): Ephemeral channel for transient progression updates.
3. Webhook Delivery Log: Audit log capturing every delivery attempt, HTTP status code, and retry timestamp.

04 Polling Mechanics and Architectural Pitfalls

Polling is the simplest approach, wherein client JavaScript polls the server at periodic intervals:

// Naive polling implementation in frontend
async function pollJobStatus(jobId: string) {
  const timer = setInterval(async () => {
    try {
      const res = await fetch(`/v1/jobs/${jobId}`);
      const data = await res.json();
      console.log("Current progress:", data.progress);
      if (data.status === "completed" || data.status === "failed") {
        clearInterval(timer); // Terminate timer on completion
      }
    } catch (err) {
      console.error("Polling error:", err);
    }
  }, 2000); // Check every 2 seconds
}

05 Server-Sent Events (SSE) Protocol Mechanics

Server-Sent Events (SSE) is a W3C/WHATWG standard allowing a server to push continuous text event streams to web clients over a persistent standard HTTP connection.

SSE Wire Protocol Framing Specifications:

• id: <identifier> — Monotonic event identifier. If severed, browsers pass this value in the Last-Event-ID header upon reconnecting.
• event: <name> — Event type tag (e.g., token, progress, completed, error).
• data: <payload> — Serialized event payload (typically JSON strings).
• (Double Newline) — Mandatory boundary delimiter terminating the frame.

06 WebSocket Protocol and Bidirectional Channels

WebSockets upgrade an HTTP handshake into a persistent TCP socket (ws:// or wss://), facilitating full-duplex framing where both client and server transmit payloads simultaneously.

Why is SSE Preferred over WebSockets for Conversational AI?

• SSE operates seamlessly over standard HTTP/2 and HTTP/3 without proxy blocking or corporate firewall drops.
• Browsers provide automatic reconnection and event tracking natively via EventSource.
• Eliminates complex socket state clusters, distributed ping/pong heartbeats, and sticky session overhead.

07 Architectural Decision Framework: Which Primitive to Choose?

Apply the following engineering criteria to select the appropriate real-time transport:

Never adopt WebSockets purely for novelty. Choose the simplest transport primitive capable of satisfying system reliability requirements.

08 Part B: Production Project Structure

To integrate real-time streaming and webhooks into our backend, we structure the workspace into clean modules:

09 Building a Demo SSE Endpoint with FastAPI

We begin by implementing an offline demo endpoint to master SSE wire framing and async generators:

import asyncio
import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI(title="Byteion Real-Time AI")

def sse_frame(event: str, data: dict, event_id: str | None = None) -> str:
    """Format payload according to W3C Server-Sent Events wire spec."""
    lines = []
    if event_id is not None:
        lines.append(f"id: {event_id}")
    lines.append(f"event: {event}")
    lines.append("data: " + json.dumps(data, ensure_ascii=False))
    return "\n".join(lines) + "\n\n"

@app.get("/v1/demo/stream")
async def demo_stream():
    async def event_generator():
        yield sse_frame("progress", {"stage": "starting", "percent": 0}, event_id="1")
        sample_tokens = ["AI", "Engineering", "demands", "resilient", "real-time", "systems."]
        for idx, token in enumerate(sample_tokens, start=2):
            await asyncio.sleep(0.12) # Simulate model latency
            yield sse_frame("token", {"text": token + " "}, event_id=str(idx))
        yield sse_frame("completed", {"status": "success"}, event_id="8")

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
            "Connection": "keep-alive",
        },
    )

Execute curl -N with unbuffered terminal output to observe live stream delivery:

curl -N http://127.0.0.1:8000/v1/demo/stream

# app/services/ai_stream.py
import os
import json
import logging
from openai import AsyncOpenAI

logger = logging.getLogger(__name__)
client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY", "mock-key"))

def sse_frame(event: str, data: dict, event_id: str | None = None) -> str:
    lines = []
    if event_id is not None:
        lines.append(f"id: {event_id}")
    lines.append(f"event: {event}")
    lines.append("data: " + json.dumps(data, ensure_ascii=False))
    return "\n".join(lines) + "\n\n"

async def stream_ai_answer(prompt: str):
    sequence = 0
    try:
        # Request semantic streaming from OpenAI Responses API
        stream = await client.responses.create(
            model=os.environ.get("OPENAI_MODEL", "gpt-4o-mini"),
            input=prompt,
            stream=True,
        )

        async for event in stream:
            # Output text delta emission
            if event.type == "response.output_text.delta":
                sequence += 1
                yield sse_frame("token", {"text": event.delta}, str(sequence))
            
            # Model output complete
            elif event.type == "response.completed":
                sequence += 1
                yield sse_frame("completed", {"response_id": event.response.id, "status": "completed"}, str(sequence))
            
            # Provider stream error
            elif event.type == "error":
                sequence += 1
                yield sse_frame("error", {"code": "provider_error", "message": "Stream interrupted by provider."}, str(sequence))
                return

    except Exception as exc:
        logger.exception("AI streaming invocation failed")
        sequence += 1
        yield sse_frame("error", {"code": "server_error", "message": "Internal generation failure."}, str(sequence))

# app/api/answers.py
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from app.services.ai_stream import stream_ai_answer

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

class AnswerRequest(BaseModel):
    prompt: str = Field(min_length=1, max_length=4000)

@router.post("/stream")
async def answer_stream(payload: AnswerRequest):
    return StreamingResponse(
        stream_ai_answer(payload.prompt),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
            "Connection": "keep-alive",
        },
    )

// src/lib/sse.ts
export type ParsedSSEEvent = {
  event: string;
  data: string;
  id?: string;
};

export function parseSSEFrame(frame: string): ParsedSSEEvent | null {
  let event = "message";
  let id: string | undefined;
  const dataLines: string[] = [];

  for (const line of frame.split("\n")) {
    if (line.startsWith(":")) {
      // SSE comment / heartbeat ping
      continue;
    }
    const separator = line.indexOf(":");
    const field = separator === -1 ? line : line.slice(0, separator);
    let value = separator === -1 ? "" : line.slice(separator + 1);

    if (value.startsWith(" ")) {
      value = value.slice(1);
    }

    if (field === "event") {
      event = value;
    } else if (field === "data") {
      dataLines.push(value);
    } else if (field === "id") {
      id = value;
    }
  }

  if (dataLines.length === 0) return null;

  return {
    event,
    data: dataLines.join("\n"),
    id,
  };
}

// src/components/AIAnswer.tsx
import React, { useState, useRef } from "react";
import { parseSSEFrame } from "../lib/sse";

export function AIAnswer({ authToken }: { authToken: string }) {
  const [prompt, setPrompt] = useState("");
  const [answer, setAnswer] = useState("");
  const [status, setStatus] = useState<"idle" | "streaming" | "completed" | "failed" | "cancelled">("idle");
  const controllerRef = useRef<AbortController | null>(null);

  async function handleGenerate() {
    controllerRef.current?.abort();
    const controller = new AbortController();
    controllerRef.current = controller;

    setAnswer("");
    setStatus("streaming");

    try {
      const response = await fetch("/v1/answers/stream", {
        method: "POST",
        headers: {
          "Content-Type": "application/json",
          Authorization: `Bearer ${authToken}`,
        },
        body: JSON.stringify({ prompt }),
        signal: controller.signal,
      });

      if (!response.ok || !response.body) {
        throw new Error(`Streaming failed with status: ${response.status}`);
      }

      const reader = response.body.getReader();
      const decoder = new TextDecoder();
      let buffer = "";

      while (true) {
        const { value, done } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        buffer = buffer.replace(/\r\n/g, "\n");

        let boundary: number;
        while ((boundary = buffer.indexOf("\n\n")) !== -1) {
          const frame = buffer.slice(0, boundary);
          buffer = buffer.slice(boundary + 2);

          const parsed = parseSSEFrame(frame);
          if (!parsed) continue;

          const data = JSON.parse(parsed.data);
          if (parsed.event === "token" && data.text) {
            setAnswer((prev) => prev + data.text);
          } else if (parsed.event === "completed") {
            setStatus("completed");
          } else if (parsed.event === "error") {
            setStatus("failed");
            throw new Error(data.message || "Model error");
          }
        }
      }
    } catch (err: any) {
      if (controller.signal.aborted) {
        setStatus("cancelled");
      } else {
        setStatus("failed");
      }
    }
  }

  return (
    <div className="rounded-xl border border-border p-4 bg-card">
      <textarea
        value={prompt}
        onChange={(e) => setPrompt(e.target.value)}
        className="w-full rounded-lg border border-border p-2 text-xs bg-background"
        placeholder="Ask a question..."
      />
      <div className="flex gap-2 mt-2">
        <button
          onClick={handleGenerate}
          disabled={status === "streaming" || !prompt.trim()}
          className="bg-cyan-600 text-white px-3 py-1.5 rounded-lg text-xs font-bold"
        >
          Generate
        </button>
        <button
          onClick={() => controllerRef.current?.abort()}
          disabled={status !== "streaming"}
          className="border border-border text-foreground px-3 py-1.5 rounded-lg text-xs"
        >
          Stop
        </button>
      </div>
      <div className="mt-3 text-xs text-muted-foreground">Status: {status}</div>
      <div className="mt-2 p-3 bg-muted/30 rounded-lg text-xs font-mono whitespace-pre-wrap">{answer || "..."}</div>
    </div>
  );
}

-- Authoritative business job record
CREATE TABLE ai_jobs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL,
    created_by UUID NOT NULL,
    status VARCHAR(30) NOT NULL DEFAULT 'queued', -- 'queued', 'running', 'completed', 'failed'
    stage VARCHAR(100),
    progress SMALLINT NOT NULL DEFAULT 0,
    result JSONB,
    error_code VARCHAR(100),
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Durable incremental event log supporting replay
CREATE TABLE ai_job_events (
    job_id UUID NOT NULL REFERENCES ai_jobs(id) ON DELETE CASCADE,
    event_sequence BIGINT NOT NULL,
    event_type VARCHAR(40) NOT NULL, -- 'progress', 'token', 'tool_status', 'completed'
    payload JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (job_id, event_sequence)
);

CREATE INDEX idx_ai_job_events_replay 
ON ai_job_events (job_id, event_sequence ASC);

# app/schemas/events.py
from pydantic import BaseModel, Field

class ProgressEventPayload(BaseModel):
    job_id: str
    stage: str
    percent: int = Field(ge=0, le=100)
    message: str | None = None

class TokenEventPayload(BaseModel):
    run_id: str
    text: str

class CompletedEventPayload(BaseModel):
    job_id: str
    status: str = "completed"
    result_url: str

class ErrorEventPayload(BaseModel):
    job_id: str
    code: str
    message: str
    retryable: bool = False

CREATE TABLE webhook_deliveries (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id UUID NOT NULL,
    event_id UUID NOT NULL,
    endpoint_url TEXT NOT NULL,
    attempt_count INTEGER NOT NULL DEFAULT 0,
    status VARCHAR(30) NOT NULL DEFAULT 'pending', -- 'pending', 'succeeded', 'retrying', 'dead_letter'
    next_attempt_at TIMESTAMPTZ,
    last_http_status INTEGER,
    last_error TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_webhook_deliveries_pending 
ON webhook_deliveries (status, next_attempt_at) 
WHERE status IN ('pending', 'retrying');

# app/security/webhook_signatures.py
import hmac
import hashlib
import time

def sign_webhook_payload(secret: str, raw_body: bytes) -> tuple[int, str]:
    """
    Constructs an HMAC-SHA256 signature over timestamp and raw body bytes.
    Canonical format: timestamp.raw_body
    """
    timestamp = int(time.time())
    signed_payload = str(timestamp).encode("utf-8") + b"." + raw_body
    digest = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).hexdigest()
    return timestamp, digest

def verify_webhook_signature(
    secret: str,
    raw_body: bytes,
    signature_header: str,
    timestamp_header: int,
    tolerance_seconds: int = 300,
) -> bool:
    """
    Verifies signature using constant-time comparison and enforces timestamp tolerance window.
    """
    now = int(time.time())
    if abs(now - timestamp_header) > tolerance_seconds:
        return False # Mitigate replay attacks outside 5-minute window

    signed_payload = str(timestamp_header).encode("utf-8") + b"." + raw_body
    expected_digest = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).hexdigest()
    
    # Constant-time comparison prevents timing attacks
    return hmac.compare_digest(expected_digest, signature_header)

# app/services/agent_stream.py
from agents import Agent, Runner
from openai.types.responses import ResponseTextDeltaEvent

async def stream_agent_execution(prompt: str):
    agent = Agent(
        name="Enterprise Financial Auditor",
        instructions="Analyze financial risks and use compliance tools when required.",
    )

    result = Runner.run_streamed(agent, input=prompt)

    async for event in result.stream_events():
        # Case 1: Live raw text delta
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            yield f"event: token\ndata: {json.dumps({'text': event.data.delta})}\n\n"
        
        # Case 2: Multi-agent handoff event
        elif event.type == "agent_updated_stream_event":
            yield f"event: tool_status\ndata: {json.dumps({'stage': f'Switched agent to: {event.new_agent.name}'})}\n\n"

    if result.is_complete:
        yield f"event: completed\ndata: {json.dumps({'status': 'completed'})}\n\n"