ওয়েবহুক ও সার্ভার-সেন্ট ইভেন্ট দিয়ে স্ট্রিমিং

mkdir -p ~/projects/webhooks-demo
cd ~/projects/webhooks-demo
uv init --bare
uv add -q "fastapi[standard]" requests httpx
# uv-এর বদলে pip দিয়ে:
#   python -m venv .venv && source .venv/bin/activate
#   pip install -q "fastapi[standard]" requests httpx

Initialized project `webhooks-demo`

# file: signing.py
import hashlib
import hmac


class SignatureError(Exception):
    def __init__(self, message: str, status: int = 401):
        super().__init__(message)
        self.status = status


def sign(secret: str, timestamp: int, body: bytes) -> str:
    message = str(timestamp).encode() + b"." + body
    return "sha256=" + hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()


def verify_signature(secret: str, timestamp: str | None, signature: str | None,
                     body: bytes, now: int, tolerance: int = 300) -> None:
    if not timestamp or not signature:
        raise SignatureError("Missing signature headers", 400)
    try:
        sent_at = int(timestamp)
    except ValueError:
        raise SignatureError("Invalid timestamp", 400)
    if abs(now - sent_at) > tolerance:
        raise SignatureError("Timestamp outside the allowed window", 400)
    expected = sign(secret, sent_at, body)
    if not hmac.compare_digest(expected.encode(), signature.encode()):
        raise SignatureError("Signature does not match", 401)

import json
from signing import SignatureError, sign, verify_signature

SECRET = "whsec_demo"
TIMESTAMP = 1767225600
body = b'{"id":"evt_0001","type":"order.paid"}'
signature = sign(SECRET, TIMESTAMP, body)
print(signature)

def check(label, secret=SECRET, ts=TIMESTAMP, sig=signature, payload=body, now=TIMESTAMP + 10):
    try:
        verify_signature(secret, str(ts), sig, payload, now)
        print(f"{label}: accepted")
    except SignatureError as exc:
        print(f"{label}: rejected {exc.status} ({exc})")

check("untouched")
check("body changed", payload=b'{"id":"evt_0001","type":"order.refunded"}')
check("wrong secret", secret="whsec_other")
check("an hour later", now=TIMESTAMP + 3600)
check("new timestamp, old signature", ts=TIMESTAMP + 3600, now=TIMESTAMP + 3600)
check("re-serialised JSON", payload=json.dumps(json.loads(body)).encode())
check("no headers", ts="", sig="")

sha256=ecb28056ad4c9ecb70a48f76e50c5fb959fea17833ae3ce76e4fc58efc825916
untouched: accepted
body changed: rejected 401 (Signature does not match)
wrong secret: rejected 401 (Signature does not match)
an hour later: rejected 400 (Timestamp outside the allowed window)
new timestamp, old signature: rejected 401 (Signature does not match)
re-serialised JSON: rejected 401 (Signature does not match)
no headers: rejected 400 (Missing signature headers)

# file: receiver.py
import json
import os
import time

from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from pydantic import BaseModel

from signing import SignatureError, verify_signature

app = FastAPI(title="Webhook receiver")
SECRET = os.environ["WEBHOOK_SECRET"]
SEEN: set[str] = set()
PROCESSED: list[str] = []
OUTAGE = {"remaining": 0}


class Outage(BaseModel):
    count: int


def handle(event: dict) -> None:
    PROCESSED.append(event["id"])


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


@app.post("/webhooks/orders")
async def receive(request: Request, background: BackgroundTasks):
    body = await request.body()
    try:
        verify_signature(
            SECRET,
            request.headers.get("x-webhook-timestamp"),
            request.headers.get("x-webhook-signature"),
            body,
            now=int(time.time()),
        )
    except SignatureError as exc:
        raise HTTPException(exc.status, str(exc))
    if OUTAGE["remaining"] > 0:
        OUTAGE["remaining"] -= 1
        raise HTTPException(503, "Temporarily unavailable")
    event = json.loads(body)
    if event["id"] in SEEN:
        return {"status": "duplicate"}
    SEEN.add(event["id"])
    background.add_task(handle, event)
    return {"status": "accepted"}


@app.post("/admin/fail")
def fail_next(outage: Outage):
    OUTAGE["remaining"] = outage.count
    return OUTAGE


@app.get("/processed")
def processed():
    return {"processed": PROCESSED}

# file: sender.py
import itertools
import json
import time

import requests
from fastapi import BackgroundTasks, FastAPI
from pydantic import BaseModel

from signing import sign

app = FastAPI(title="Webhook sender")
SUBSCRIPTIONS: list[dict] = []
DELIVERIES: list[dict] = []
EVENT_IDS = itertools.count(1)
RETRY_DELAYS = [0.05, 0.1, 0.2]


class SubscriptionIn(BaseModel):
    url: str
    secret: str


def deliver(event: dict, subscription: dict) -> None:
    body = json.dumps(event, separators=(",", ":"), sort_keys=True).encode()
    for attempt in range(1, len(RETRY_DELAYS) + 2):
        timestamp = int(time.time())
        headers = {
            "Content-Type": "application/json",
            "X-Webhook-Id": event["id"],
            "X-Webhook-Timestamp": str(timestamp),
            "X-Webhook-Signature": sign(subscription["secret"], timestamp, body),
        }
        try:
            status = requests.post(subscription["url"], data=body, headers=headers, timeout=(2, 5)).status_code
        except requests.RequestException:
            status = None
        DELIVERIES.append({"event_id": event["id"], "attempt": attempt, "status": status})
        if status is not None and 200 <= status < 300:
            return
        if attempt <= len(RETRY_DELAYS):
            time.sleep(RETRY_DELAYS[attempt - 1])


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


@app.post("/subscriptions", status_code=201)
def subscribe(subscription: SubscriptionIn):
    SUBSCRIPTIONS.append(subscription.model_dump())
    return {"subscriptions": len(SUBSCRIPTIONS)}


@app.post("/orders/{order_id}/pay")
def pay(order_id: int, background: BackgroundTasks):
    event = {"id": f"evt_{next(EVENT_IDS):04d}", "type": "order.paid",
             "data": {"order_id": order_id, "amount": 25000}}
    for subscription in SUBSCRIPTIONS:
        background.add_task(deliver, event, subscription)
    return {"event_id": event["id"]}


@app.get("/deliveries")
def deliveries():
    return DELIVERIES

export WEBHOOK_SECRET=whsec_demo
uv run fastapi run receiver.py --port 8001 > receiver.log 2>&1 &
echo $! > receiver.pid
uv run fastapi run sender.py --port 8000 > sender.log 2>&1 &
echo $! > sender.pid
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8001/health; echo
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health; echo

{"status":"ok"}
{"status":"ok"}

import requests, time

SENDER, RECEIVER = "http://localhost:8000", "http://localhost:8001"

print(requests.post(f"{SENDER}/subscriptions",
                    json={"url": f"{RECEIVER}/webhooks/orders", "secret": "whsec_demo"}).json())
print(requests.post(f"{RECEIVER}/admin/fail", json={"count": 2}).json())
print(requests.post(f"{SENDER}/orders/1001/pay").json())

def wait_until(check, timeout=5.0):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        value = check()
        if value:
            return value
        time.sleep(0.05)
    raise TimeoutError("condition never became true")

deliveries = wait_until(lambda: (d := requests.get(f"{SENDER}/deliveries").json()) and d[-1]["status"] == 200 and d)
for delivery in deliveries:
    print(delivery)
print(requests.get(f"{RECEIVER}/processed").json())

{'subscriptions': 1}
{'remaining': 2}
{'event_id': 'evt_0001'}
{'event_id': 'evt_0001', 'attempt': 1, 'status': 503}
{'event_id': 'evt_0001', 'attempt': 2, 'status': 503}
{'event_id': 'evt_0001', 'attempt': 3, 'status': 200}
{'processed': ['evt_0001']}

from signing import sign
import json, time

def send(event, secret="whsec_demo", age=0, signature=None, timestamp=None):
    body = json.dumps(event, separators=(",", ":"), sort_keys=True).encode()
    ts = timestamp if timestamp is not None else int(time.time()) - age
    headers = {"X-Webhook-Timestamp": str(ts),
               "X-Webhook-Signature": signature or sign(secret, ts, body)}
    response = requests.post(f"{RECEIVER}/webhooks/orders", data=body, headers=headers)
    return response.status_code, response.json()

event = {"id": "evt_9001", "type": "order.paid", "data": {"order_id": 7, "amount": 100}}
print("first delivery   ", send(event))
print("same event again ", send(event))
print("recorded last hr ", send({**event, "id": "evt_9002"}, age=3600))
print("wrong secret     ", send({**event, "id": "evt_9003"}, secret="whsec_guess"))

genuine = int(time.time())
old_signature = sign("whsec_demo", genuine - 3600, json.dumps(event, separators=(",", ":"), sort_keys=True).encode())
print("stolen signature ", send(event, signature=old_signature, timestamp=genuine))
print(requests.get(f"{RECEIVER}/processed").json())

first delivery    (200, {'status': 'accepted'})
same event again  (200, {'status': 'duplicate'})
recorded last hr  (400, {'detail': 'Timestamp outside the allowed window'})
wrong secret      (401, {'detail': 'Signature does not match'})
stolen signature  (401, {'detail': 'Signature does not match'})
{'processed': ['evt_0001', 'evt_9001']}

kill $(cat receiver.pid) $(cat sender.pid)
while curl -s -o /dev/null http://localhost:8000/health; do sleep 0.1; done
echo stopped

stopped

# file: main.py
import asyncio
from collections.abc import AsyncIterable
from typing import Annotated

from fastapi import FastAPI, Header
from fastapi.responses import StreamingResponse
from fastapi.sse import EventSourceResponse, ServerSentEvent

app = FastAPI(title="Streaming demo")
WORDS = "Server-Sent Events are plain text over ordinary HTTP".split()
CLOSED: list[str] = []


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


@app.get("/plain")
async def plain():
    async def chunks():
        for word in WORDS:
            await asyncio.sleep(0.01)
            yield word + " "

    return StreamingResponse(chunks(), media_type="text/plain")


@app.get("/stream", response_class=EventSourceResponse)
async def stream(
    last_event_id: Annotated[int | None, Header()] = None,
    fail_at: int | None = None,
) -> AsyncIterable[ServerSentEvent]:
    start = last_event_id or 0
    yield ServerSentEvent(comment="stream open", retry=3000)
    for number, word in enumerate(WORDS[start:], start=start + 1):
        await asyncio.sleep(0.02)
        if number == fail_at:
            yield ServerSentEvent(data={"message": "the model failed"}, event="error")
            return
        yield ServerSentEvent(data={"token": word}, event="token", id=str(number))
    yield ServerSentEvent(data={"done": True}, event="done")


@app.get("/ticks", response_class=EventSourceResponse)
async def ticks() -> AsyncIterable[ServerSentEvent]:
    try:
        for number in range(1, 1000):
            await asyncio.sleep(0.02)
            yield ServerSentEvent(data={"tick": number})
    finally:
        CLOSED.append("ticks")


@app.get("/closed")
def closed():
    return {"closed": len(CLOSED)}

uv run fastapi run main.py --port 8000 > server.log 2>&1 &
echo $! > server.pid
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health; echo
curl -sN http://localhost:8000/plain; echo

{"status":"ok"}
Server-Sent Events are plain text over ordinary HTTP

curl -sN -X POST http://localhost:8800/v1/chat \
  -H "Authorization: Bearer token-nadia-001" \
  -H "Content-Type: application/json" \
  -d '{"messages": [{"role": "user", "content": "What is RAG?"}], "stream": true}'

data: {"delta": "This "}

data: {"delta": "is "}

data: {"delta": "a "}

data: {"delta": "practice "}

data: {"delta": "answer "}

data: {"delta": "about: "}

data: {"delta": "What "}

data: {"delta": "is "}

data: {"delta": "RAG. "}

data: {"delta": "A "}

data: {"delta": "real "}

data: {"delta": "model "}

data: {"delta": "would "}

data: {"delta": "say "}

data: {"delta": "more. "}

data: {"done": true, "usage": {"prompt_tokens": 3, "completion_tokens": 15, "total_tokens": 18}}

curl -sN http://localhost:8000/stream

: stream open
retry: 3000

event: token
data: {"token": "Server-Sent"}
id: 1

event: token
data: {"token": "Events"}
id: 2

event: token
data: {"token": "are"}
id: 3

event: token
data: {"token": "plain"}
id: 4

event: token
data: {"token": "text"}
id: 5

event: token
data: {"token": "over"}
id: 6

event: token
data: {"token": "ordinary"}
id: 7

event: token
data: {"token": "HTTP"}
id: 8

event: done
data: {"done": true}

curl -sN -H "Last-Event-ID: 5" http://localhost:8000/stream | grep -E "^(id|event: done)"

id: 6
id: 7
id: 8
event: done

def parse_sse(lines):
    """Turn the lines of an SSE response into events: dicts with event, data and id."""
    event = {"event": "message", "data": [], "id": None}
    for line in lines:
        if line == "":
            if event["data"]:
                yield {**event, "data": "\n".join(event["data"])}
            event = {"event": "message", "data": [], "id": None}
        elif line.startswith(":"):
            continue
        else:
            field, _, value = line.partition(":")
            value = value.removeprefix(" ")
            if field == "data":
                event["data"].append(value)
            elif field in ("event", "id"):
                event[field] = value

CHAT = "http://localhost:8800/v1/chat"
AUTH = {"Authorization": "Bearer token-nadia-001"}
question = {"messages": [{"role": "user", "content": "What is retrieval augmented generation?"}], "stream": True}

pieces, usage = [], None
with requests.post(CHAT, json=question, headers=AUTH, stream=True, timeout=(3, 30)) as response:
    response.raise_for_status()
    response.encoding = "utf-8"
    for event in parse_sse(response.iter_lines(decode_unicode=True, chunk_size=None)):
        data = json.loads(event["data"])
        if data.get("done"):
            usage = data["usage"]
        else:
            pieces.append(data["delta"])

print(len(pieces), "pieces; first three:", pieces[:3])
print("".join(pieces).strip())
print(usage)

17 pieces; first three: ['This ', 'is ', 'a ']
This is a practice answer about: What is retrieval augmented generation. A real model would say more.
{'prompt_tokens': 5, 'completion_tokens': 17, 'total_tokens': 22}

import httpx

with httpx.stream("POST", CHAT, json=question, headers=AUTH, timeout=httpx.Timeout(30.0, connect=3.0)) as response:
    response.raise_for_status()
    text = "".join(json.loads(e["data"]).get("delta", "") for e in parse_sse(response.iter_lines()))
print(text.strip())

This is a practice answer about: What is retrieval augmented generation. A real model would say more.

APP = "http://localhost:8000"

def read_tokens(stop_after=None, last_event_id=None, params=None):
    headers = {"Last-Event-ID": last_event_id} if last_event_id else {}
    got = []
    with requests.get(f"{APP}/stream", params=params, headers=headers, stream=True, timeout=5) as response:
        response.encoding = "utf-8"
        for event in parse_sse(response.iter_lines(decode_unicode=True, chunk_size=None)):
            if event["event"] == "token":
                got.append((event["id"], json.loads(event["data"])["token"]))
                if stop_after and len(got) >= stop_after:
                    break
            elif event["event"] == "error":
                got.append(("error", json.loads(event["data"])["message"]))
    return got

first = read_tokens(stop_after=3)
print("connection dropped after:", first)
rest = read_tokens(last_event_id=first[-1][0])
print("resumed:", rest)
print("whole sentence:", " ".join(word for _, word in first + rest))
print("model failure:", read_tokens(params={"fail_at": 3}))

connection dropped after: [('1', 'Server-Sent'), ('2', 'Events'), ('3', 'are')]
resumed: [('4', 'plain'), ('5', 'text'), ('6', 'over'), ('7', 'ordinary'), ('8', 'HTTP')]
whole sentence: Server-Sent Events are plain text over ordinary HTTP
model failure: [('1', 'Server-Sent'), ('2', 'Events'), ('error', 'the model failed')]

with requests.get(f"{APP}/ticks", stream=True, timeout=5) as response:
    response.encoding = "utf-8"
    for count, event in enumerate(parse_sse(response.iter_lines(decode_unicode=True, chunk_size=None)), start=1):
        print(event["data"])
        if count == 3:
            break

print("server noticed:", wait_until(lambda: requests.get(f"{APP}/closed").json()["closed"] or None))

{"tick": 1}
{"tick": 2}
{"tick": 3}
server noticed: 1

# 1. Easy
exercise_body = b'{"id":"evt_9"}'
exercise_signature = sign("whsec_exercise", 1767225600, exercise_body)
print(exercise_signature)


# 2. Medium
def try_verify(payload, now):
    try:
        verify_signature("whsec_exercise", "1767225600", exercise_signature, payload, now)
        return "accepted"
    except SignatureError as exc:
        return f"rejected: {exc}"

print(try_verify(exercise_body, 1767225610))
print(try_verify(b'{"id":"evt_8"}', 1767225610))
print(try_verify(exercise_body, 1767229600))


# 3. Medium
payload = {"messages": [{"role": "user", "content": "Why use webhooks?"}], "stream": True}
deltas, total = [], None
with httpx.stream("POST", CHAT, json=payload, headers=AUTH, timeout=30.0) as response:
    response.raise_for_status()
    for event in parse_sse(response.iter_lines()):
        data = json.loads(event["data"])
        if data.get("done"):
            total = data["usage"]["total_tokens"]
        else:
            deltas.append(data["delta"])
print(len(deltas), "pieces")
print("".join(deltas).strip())
print("total_tokens:", total)


# 4. Hard
def read_until_done(drop_after):
    tokens, last_id, connections = [], None, 0
    while True:
        connections += 1
        headers = {"Last-Event-ID": last_id} if last_id else {}
        seen, finished = 0, False
        with requests.get(f"{APP}/stream", headers=headers, stream=True, timeout=5) as response:
            response.encoding = "utf-8"
            for event in parse_sse(response.iter_lines(decode_unicode=True, chunk_size=None)):
                if event["event"] == "done":
                    finished = True
                    break
                if event["event"] == "token":
                    tokens.append(json.loads(event["data"])["token"])
                    last_id = event["id"]
                    seen += 1
                    if connections == 1 and seen == drop_after:
                        break
        if finished:
            return " ".join(tokens), connections

print(read_until_done(drop_after=4))

sha256=06ad48768c3077908ecc479f6b333b30dec6e25fe743f7490bbad8a40e7d4643
accepted
rejected: Signature does not match
rejected: Timestamp outside the allowed window
15 pieces
This is a practice answer about: Why use webhooks. A real model would say more.
total_tokens: 18
('Server-Sent Events are plain text over ordinary HTTP', 2)

kill $(cat server.pid)