Webhook-Driven Dynamic Re-Routing

Static routes go stale the moment a road closes, a delivery window slips, or a new order lands mid-shift, and polling every vehicle’s route on a timer wastes API quota and still reacts late. This section of the Routing API Automation & Fleet Integration guide covers the alternative: an event-driven service that receives webhook notifications the instant something changes, verifies and deduplicates them, recomputes the affected route against a routing engine, and pushes the result to the vehicle without the fleet system ever asking. Two focused walkthroughs build on the architecture here — building a FastAPI re-route webhook drills into the endpoint mechanics, and traffic-incident-triggered re-routing applies the pattern to geofenced road closures. This page covers the end-to-end service: schema, security, idempotency, recomputation, and delivery.

Prerequisites

The service is a thin, stateless FastAPI layer in front of a durable queue, an idempotency store, and whichever routing engine already serves your fleet. Confirm these before wiring the webhook.

System requirements

Python environment

# Install server-side dependencies (Python 3.9+)
pip install "fastapi[all]" uvicorn httpx redis pydantic-settings

External dependencies

If you have not deployed a local routing engine yet, start with deploying OSRM with Docker for local routing — the recompute step below assumes an HTTP endpoint with the standard /route/v1 interface.

Conceptual Architecture

Webhook delivery is inherently at-least-once: network retries, load balancer timeouts, and naive retry logic on the sender’s side all mean your endpoint will receive the same event more than once. The architecture above treats that as a first-class constraint rather than an edge case, and separates four concerns that are easy to conflate into a single slow handler:

Ingestion and authentication. The endpoint accepts a POST, verifies an HMAC-SHA256 signature over the raw request body, and rejects anything that fails the check before touching business logic. This keeps unauthenticated senders from triggering expensive recomputes.

Deduplication. A Redis-backed idempotency key — derived from the event’s own identifier, not from request metadata — guarantees that a retried delivery is a no-op. This is what makes at-least-once delivery safe to treat as effectively-once processing downstream.

Asynchronous recomputation. The endpoint acknowledges within milliseconds (202 Accepted) and hands the actual route recalculation to a background task or worker queue. Webhook senders typically time out and retry after a few seconds; recomputing against a routing engine under load can take longer than that, so the acknowledgment and the work must be decoupled.

Delivery with acknowledgment. Pushing the new route to a moving vehicle is itself unreliable — connections drop, devices sleep, cellular coverage lapses. The service tracks whether the device acknowledged receipt and retries with backoff until it does or a maximum attempt count is reached, rather than firing once and assuming success.

This decomposition maps directly onto the sequence diagram: verification and idempotency happen synchronously inside the request/response cycle; recomputation and push happen afterward, off the critical path, with their own retry semantics. The traffic-incident-triggered re-routing page shows the same architecture applied to a geofenced incident feed, where the “affected vehicles” step replaces a single vehicle_id with a spatial join against active routes.

Step-by-Step Implementation

1. Define the Webhook Event Schema

Model the event with Pydantic so malformed payloads fail validation before reaching business logic. Keep the schema permissive enough to cover multiple trigger types with one endpoint.

# requires: fastapi, pydantic (pip install "fastapi[all]")
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from pydantic import BaseModel, Field, field_validator


class RerouteEventType(str, Enum):
    TRAFFIC_INCIDENT = "traffic_incident"
    ORDER_ADDED = "order_added"
    ROAD_CLOSURE = "road_closure"
    ETA_SLIP = "eta_slip"


class RerouteEvent(BaseModel):
    event_id: str = Field(..., description="Source-assigned unique ID; used as the idempotency key")
    event_type: RerouteEventType
    vehicle_id: str
    current_lat: float
    current_lon: float
    affected_edge_ids: Optional[list[str]] = None
    remaining_stops: list[dict] = Field(default_factory=list)
    occurred_at: datetime

    @field_validator("occurred_at")
    @classmethod
    def must_be_recent(cls, v: datetime) -> datetime:
        age_s = (datetime.now(timezone.utc) - v).total_seconds()
        if age_s > 300:
            raise ValueError(f"event is {age_s:.0f}s stale; reject to avoid routing on outdated conditions")
        return v

Rejecting stale events at the schema level — anything older than a few minutes — prevents a backlogged webhook queue from triggering routes based on conditions that no longer hold.

2. Verify HMAC Signatures on Incoming Requests

Signature verification must run against the exact raw bytes the sender signed, before FastAPI’s body-parsing machinery re-serializes anything. Read the body manually inside a dependency:

# requires: fastapi (pip install "fastapi[all]")
import hashlib
import hmac
import os

from fastapi import Depends, HTTPException, Request

WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"].encode()


async def verify_signature(request: Request) -> bytes:
    """Validate the X-Signature header against the raw request body."""
    raw_body = await request.body()
    signature = request.headers.get("X-Signature", "")

    expected = hmac.new(WEBHOOK_SECRET, raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(f"sha256={expected}", signature):
        raise HTTPException(status_code=401, detail="Invalid webhook signature")

    return raw_body

hmac.compare_digest runs in constant time regardless of where the strings first differ, which prevents an attacker from timing the comparison to guess the correct signature byte by byte. Passing raw_body back out of the dependency lets the route handler re-parse it into the RerouteEvent model without a second body read, which FastAPI does not support by default.

3. Enforce Idempotency with Redis

Use a single atomic SET ... NX EX call — not a separate existence check followed by a write — so concurrent duplicate deliveries cannot both pass the check.

# requires: redis (pip install redis)
import redis.asyncio as redis

r = redis.from_url("redis://localhost:6379/0", decode_responses=True)

IDEMPOTENCY_TTL_S = 3600


async def claim_event(event_id: str) -> bool:
    """Return True if this is the first delivery of event_id, False if it's a duplicate."""
    was_set = await r.set(f"reroute:seen:{event_id}", "1", nx=True, ex=IDEMPOTENCY_TTL_S)
    return bool(was_set)

nx=True makes the write conditional on the key not already existing, and Redis executes the check-and-set as one operation, closing the race window a two-step check-then-set pattern leaves open under bursty webhook delivery.

4. Queue the Recompute as a Background Task

Acknowledge the webhook immediately and defer the actual recomputation. BackgroundTasks is fine for a low-volume prototype; the Production Optimization section below covers moving this to Celery or RQ for durability.

# requires: fastapi, redis (pip install "fastapi[all]" redis)
from fastapi import BackgroundTasks, FastAPI, Depends
import json

app = FastAPI()


@app.post("/webhooks/reroute", status_code=202)
async def receive_reroute_event(
    background_tasks: BackgroundTasks,
    raw_body: bytes = Depends(verify_signature),
):
    payload = json.loads(raw_body)
    event = RerouteEvent(**payload)

    if not await claim_event(event.event_id):
        return {"status": "duplicate", "event_id": event.event_id}

    background_tasks.add_task(recompute_and_push, event)
    return {"status": "accepted", "event_id": event.event_id}

Returning 202 Accepted before the recompute finishes is the point: most webhook senders time out and retry a slow endpoint, which would otherwise turn a single event into a storm of duplicate recomputations even with idempotency in place.

5. Recompute the Route Against the Engine

Call the routing engine asynchronously with a bounded concurrency limit and a short retry policy — an engine under load should degrade gracefully rather than pile up unbounded requests.

# requires: httpx (pip install httpx)
import asyncio
import httpx

ROUTING_ENGINE_URL = "http://osrm-prod:5000"
_recompute_semaphore = asyncio.Semaphore(20)


async def recompute_route(event: RerouteEvent) -> dict:
    """Fetch an updated route from OSRM for the vehicle's current position and remaining stops."""
    coords = ";".join(
        f"{event.current_lon},{event.current_lat}",
    ) + "".join(f";{s['lon']},{s['lat']}" for s in event.remaining_stops)

    url = f"{ROUTING_ENGINE_URL}/route/v1/driving/{coords}?geometries=geojson&overview=full"

    async with _recompute_semaphore:
        async with httpx.AsyncClient(timeout=5.0) as client:
            for attempt in range(3):
                try:
                    resp = await client.get(url)
                    resp.raise_for_status()
                    data = resp.json()
                    if data["code"] == "Ok":
                        return data["routes"][0]
                except (httpx.TimeoutException, httpx.HTTPStatusError):
                    if attempt == 2:
                        raise
                    await asyncio.sleep(0.5 * (2 ** attempt))
    raise RuntimeError(f"No route found for vehicle {event.vehicle_id}")

The semaphore caps how many recomputations run concurrently against the engine regardless of how many webhook events arrive in a burst — see custom cost functions for routing solvers if the recompute needs to apply vehicle-specific weighting rather than a stock driving profile, and integrating custom traffic weights into OSRM if the event itself should also push a live weight overlay before the recompute runs.

6. Push the Updated Route to the Fleet Device

Track live connections by vehicle_id so a push can find the right device, and require an application-level acknowledgment rather than trusting the transport alone.

# requires: fastapi (pip install "fastapi[all]")
from fastapi import WebSocket, WebSocketDisconnect
import asyncio
import json

PUSH_RETRY_MAX = 4
PUSH_ACK_TIMEOUT_S = 8

connections: dict[str, WebSocket] = {}
pending_acks: dict[str, asyncio.Event] = {}


@app.websocket("/ws/vehicle/{vehicle_id}")
async def vehicle_socket(websocket: WebSocket, vehicle_id: str):
    await websocket.accept()
    connections[vehicle_id] = websocket
    try:
        while True:
            msg = json.loads(await websocket.receive_text())
            if msg.get("type") == "ack" and msg.get("event_id") in pending_acks:
                pending_acks[msg["event_id"]].set()
    except WebSocketDisconnect:
        connections.pop(vehicle_id, None)


async def push_route(vehicle_id: str, event_id: str, route: dict) -> bool:
    """Push a route update and retry with backoff until an ack or the attempt cap is reached."""
    ack_event = asyncio.Event()
    pending_acks[event_id] = ack_event

    for attempt in range(PUSH_RETRY_MAX):
        ws = connections.get(vehicle_id)
        if ws is not None:
            await ws.send_text(json.dumps({"type": "route_update", "event_id": event_id, "route": route}))
            try:
                await asyncio.wait_for(ack_event.wait(), timeout=PUSH_ACK_TIMEOUT_S)
                pending_acks.pop(event_id, None)
                return True
            except asyncio.TimeoutError:
                pass
        await asyncio.sleep(min(2 ** attempt, 10))

    pending_acks.pop(event_id, None)
    return False


async def recompute_and_push(event: RerouteEvent) -> None:
    route = await recompute_route(event)
    delivered = await push_route(event.vehicle_id, event.event_id, route)
    if not delivered:
        # See Production Optimization: route undelivered events to a dead-letter queue for replay
        pass

An MQTT-based fleet integration follows the same shape — publish to a per-vehicle topic with QoS 1, and treat the broker’s PUBACK plus an application-level ack from the device as the two conditions the retry loop waits on.

Configuration Reference

Production Optimization and Scaling

Move recomputation off in-process background tasks

BackgroundTasks runs inside the same worker process and the same event loop as the request handler; a deploy, an OOM kill, or a worker restart silently drops any task that had not finished. Replace it with a durable queue once volume justifies the operational overhead:

# requires: celery, redis (pip install celery redis)
from celery import Celery

celery_app = Celery("reroute", broker="redis://localhost:6379/1", backend="redis://localhost:6379/2")


@celery_app.task(bind=True, max_retries=3, default_retry_delay=5)
def recompute_and_push_task(self, event_payload: dict):
    import asyncio
    event = RerouteEvent(**event_payload)
    try:
        asyncio.run(recompute_and_push(event))
    except Exception as exc:
        raise self.retry(exc=exc)

Enqueue with recompute_and_push_task.delay(event.model_dump(mode="json")) from the endpoint instead of background_tasks.add_task. Celery persists the task in Redis (or SQS/RabbitMQ) until a worker picks it up, so an in-flight recompute survives an API deployment.

Scale the push layer independently

WebSocket connections are stateful and pinned to a single process, which breaks naive horizontal scaling — a push triggered on worker A cannot reach a vehicle connected to worker B. Fan out pushes through Redis pub/sub (or NATS) so any worker can trigger delivery regardless of which replica holds the connection:

# requires: redis (pip install redis)
async def publish_push(vehicle_id: str, event_id: str, route: dict) -> None:
    await r.publish(f"push:{vehicle_id}", json.dumps({"event_id": event_id, "route": route}))

Each replica subscribes to the channels for vehicles connected to it and forwards matching messages to the local WebSocket — connection ownership stays local, but the trigger becomes cluster-wide.

Protect the routing engine with a circuit breaker

A burst of incident events (a highway closure affecting fifty vehicles at once) can overwhelm the routing engine faster than the semaphore alone prevents, since queued requests still pile up behind it. Wrap recompute_route in a circuit breaker that trips after a run of failures and fails fast for a cooldown window, giving the engine room to recover instead of compounding the overload with retries.

Dead-letter undeliverable pushes

When push_route exhausts its retries, write the event and route to a dead-letter store (a Redis list or a database table) rather than discarding it. A periodic sweep can replay pending routes to any vehicle that reconnects, so a device that was briefly offline still receives its last computed route instead of driving on stale directions until the next event fires.

Rate-limit per event source

If multiple upstream systems (a telematics provider, an internal dispatch tool, a third-party traffic feed) all call the same webhook endpoint, apply a per-source token bucket keyed on an API key or client certificate. This isolates a misbehaving or retry-looping source from starving recompute capacity for the rest of the fleet.

Validation and Testing

# requires: httpx, pytest (pip install httpx pytest)
import hashlib
import hmac
import json
import time

import httpx

BASE_URL = "http://localhost:8000"
SECRET = b"test-secret"


def sign(body: bytes) -> str:
    return "sha256=" + hmac.new(SECRET, body, hashlib.sha256).hexdigest()


def make_event(event_id: str = "evt-001") -> bytes:
    return json.dumps({
        "event_id": event_id,
        "event_type": "traffic_incident",
        "vehicle_id": "veh-42",
        "current_lat": 38.8977,
        "current_lon": -77.0365,
        "remaining_stops": [{"lat": 38.8895, "lon": -77.0091}],
        "occurred_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    }).encode()


def check_valid_signature_accepted() -> None:
    body = make_event()
    resp = httpx.post(f"{BASE_URL}/webhooks/reroute", content=body, headers={"X-Signature": sign(body)})
    assert resp.status_code == 202
    assert resp.json()["status"] == "accepted"
    print("[PASS] Valid signature accepted")


def check_tampered_payload_rejected() -> None:
    body = make_event()
    tampered = body.replace(b"veh-42", b"veh-99")
    resp = httpx.post(f"{BASE_URL}/webhooks/reroute", content=tampered, headers={"X-Signature": sign(body)})
    assert resp.status_code == 401
    print("[PASS] Tampered payload rejected with 401")


def check_duplicate_delivery_deduplicated() -> None:
    body = make_event(event_id="evt-dedup-001")
    headers = {"X-Signature": sign(body)}
    first = httpx.post(f"{BASE_URL}/webhooks/reroute", content=body, headers=headers)
    second = httpx.post(f"{BASE_URL}/webhooks/reroute", content=body, headers=headers)
    assert first.json()["status"] == "accepted"
    assert second.json()["status"] == "duplicate"
    print("[PASS] Duplicate delivery deduplicated")


def check_stale_event_rejected() -> None:
    body = json.loads(make_event())
    body["occurred_at"] = "2020-01-01T00:00:00Z"
    body_bytes = json.dumps(body).encode()
    resp = httpx.post(f"{BASE_URL}/webhooks/reroute", content=body_bytes, headers={"X-Signature": sign(body_bytes)})
    assert resp.status_code == 422
    print("[PASS] Stale event rejected")


check_valid_signature_accepted()
check_tampered_payload_rejected()
check_duplicate_delivery_deduplicated()
check_stale_event_rejected()

Sanity metrics to monitor in production:

• Webhook acknowledgment latency (receive to 202) should stay under 50 ms; anything slower usually means signature verification or the idempotency check is blocking on a synchronous call.
• Recompute-to-push latency (event received to device push sent) is the number that actually matters to drivers — track it as a percentile, not an average, since routing-engine tail latency dominates under load.
• Push delivery success rate (acked within PUSH_ACK_TIMEOUT_S on the first attempt) should stay above 90%; a sustained drop points at connection-registry staleness after a deploy.

Troubleshooting

Related

• Building a FastAPI Re-Route Webhook — a closer look at the endpoint itself, including Pydantic payload validation and idempotency-key design
• Traffic-Incident-Triggered Re-Routing — applying this architecture to geofenced road closures with spatial joins against active routes
• Integrating custom traffic weights into OSRM — re-customizing the routing graph so a recompute reflects a live incident before the next full rebuild
• Custom cost functions for routing solvers — designing the objective function a recompute call should optimize for beyond raw travel time
• Async Route Matrix Automation — the batch counterpart to this event-driven pattern, for computing many-to-many travel times instead of single re-routes