require token; add relay toggle endpoints

- move websocket endpoint to {path_prefix}/websocket and reserve {path_prefix}/enable|disable
- add token-gated enable/disable routes; disabling closes active websocket and clears override paths
- client: support --token / FASTAPI_DEV_PROXY_TOKEN and inject token into relay URL
- update docs, examples, and tests for new websocket path

Author: toddsifleet

README.md

@@ -7,10 +7,18 @@ FastAPI webhook relay/proxy library for local development. Production webhooks a
 forwarded over WebSocket to a developer machine, replayed against a local server,
 and the response is sent back to production.
 
-### Status
+### Why
 
-- v0 software: unstable APIs and behavior, expect breaking changes.
-- Extracted from another project; not all original features are included yet.
+Webhook integrations are awkward to build locally because **the third-party needs to reach your machine**:
+
+- Your FastAPI server is on `localhost`, behind NAT/firewalls, and often on an unreliable dev network.
+- “Just open a port” is rarely an option (corporate networks, Wi‑Fi, security policies).
+
+Tools like ngrok (and similar tunneling/reverse-proxy setups) can solve this, but for webhook development they can also feel **overly complex**: extra moving parts, extra configuration, and yet another service to pay for.
+
+This project is the alternative I wanted: **a simple, free, open-source relay** specifically for developing webhook handlers. Instead of exposing your laptop to the internet, updating your webhook registration, you run a tiny websocket “relay” websocket in production. Then you connect a local client, receive real webhook requests, replay them against `http://localhost:...`, and send the response back upstream.
+
+A nice side-effect: you can keep the **same third-party webhook URL** (pointing at your production app) and **toggle forwarding on/off** with a config change (e.g. `enabled=False`) or by simply connecting/disconnecting the dev client—no redeploy and no “update webhook URL” dance.
 
 ### Install
 
@@ -32,39 +40,46 @@ async def sms_webhook() -> Response:
     return Response(status_code=200)
 ```
 
-Alternatively, use the class method for a one-line setup: `relay = RelayManager.for_app(app, token="secret", ...)`. Or create the relay first and call `relay.install(app)` later.
+If you prefer, you can create the relay first and call `relay.install(app)` later.
+
+When installed (default `path_prefix` is `/fastapi-dev-proxy`), the relay registers:
+
+- `{path_prefix}/websocket` (WebSocket): dev client connects here
+- `{path_prefix}/enable` (POST): enable forwarding (requires `?token=...`)
+- `{path_prefix}/disable` (POST): disable forwarding (requires `?token=...`)
 
 ### Client usage
 
-```python
-import asyncio
+The client must include the shared token as the `token` query param when connecting (e.g. `...?token=...`). To avoid hardcoding secrets in source control, read it from an environment variable.
 
-from fastapi_dev_proxy.client import run_client
+### CLI
 
-asyncio.run(
-    run_client(
-        relay_url="wss://prod.example.com/fastapi-dev-proxy?token=secret",
-        target_base_url="http://localhost:8080",
-        override_paths=["/webhook/sms", "/webhook/items/{item_id}"],
-    )
-)
 ```
+export FASTAPI_DEV_PROXY_TOKEN="your-shared-token"
 
-Override paths can include simple patterns:
-
-- `{param}` or `<param>` matches a single path segment.
-- `*` matches a single path segment.
-- `/**` at the end matches any remaining path segments.
+fastapi-dev-proxy \
+  --relay-url "wss://prod.example.com/fastapi-dev-proxy/websocket" \
+  --target-base-url http://localhost:8080 \
+  --override-path /webhook/sms \
+  --override-path /webhook/user/<uuid>
+```
 
-### CLI
+Or pass it explicitly:
 
 ```
 fastapi-dev-proxy \
-  --relay-url wss://prod.example.com/fastapi-dev-proxy?token=secret \
+  --relay-url "wss://prod.example.com/fastapi-dev-proxy/websocket" \
+  --token "your-shared-token" \
   --target-base-url http://localhost:8080 \
   --override-path /webhook/sms
 ```
 
+#### Override paths can include simple patterns:
+
+- `{param}` or `<param>` matches a single path segment.
+- `*` matches a single path segment.
+- `/**` at the end matches any remaining path segments.
+
 ### Development
 
 Install dev dependencies:
@@ -104,5 +119,3 @@ python -m pytest \
   --cov-report=xml:coverage.xml \
   --cov-report=html:htmlcov
 ```
-
-CI uploads `coverage.xml` and `htmlcov/` as workflow artifacts. If you enable Codecov for the repo, CI will also upload coverage there (badge above).

examples/basic_client.py

@@ -8,7 +8,7 @@
 def main() -> None:
     asyncio.run(
         run_client(
-            relay_url="ws://localhost:8000/fastapi-dev-proxy?token=secret",
+            relay_url="ws://localhost:8000/fastapi-dev-proxy/websocket?token=secret",
             target_base_url="http://localhost:8080",
             override_paths=["/webhook/sms"],
         )

src/fastapi_dev_proxy/client.py

@@ -6,7 +6,9 @@
 import asyncio
 import json
 import logging
+import os
 from typing import Iterable
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
 
 from httpx import AsyncClient
 from websockets import connect as connect_websocket
@@ -24,6 +26,30 @@
 
 logger = logging.getLogger(__name__)
 
+DEFAULT_TOKEN_ENV_VAR = "FASTAPI_DEV_PROXY_TOKEN"
+
+
+def _apply_token(relay_url: str, token: str | None) -> str:
+    """Return relay_url with token injected/overridden as query param."""
+    if not token:
+        return relay_url
+    parts = urlsplit(relay_url)
+    query_pairs = [(k, v) for (k, v) in parse_qsl(parts.query, keep_blank_values=True) if k != "token"]
+    query_pairs.append(("token", token))
+    new_query = urlencode(query_pairs)
+    return urlunsplit((parts.scheme, parts.netloc, parts.path, new_query, parts.fragment))
+
+
+def _resolve_token(cli_token: str | None, token_env_var: str | None) -> str:
+    if cli_token:
+        return cli_token
+    if not token_env_var:
+        raise ValueError("Token is required. Pass --token or set FASTAPI_DEV_PROXY_TOKEN.")
+    value = os.environ.get(token_env_var)
+    if value:
+        return value
+    raise ValueError(f"Token is required. Pass --token or set {token_env_var}.")
+
 
 async def _forward_request(
     client: AsyncClient,
@@ -118,6 +144,15 @@ def _parse_override_paths(values: list[str], csv_values: list[str]) -> list[str]
 def _build_parser() -> argparse.ArgumentParser:  # pragma: no cover
     parser = argparse.ArgumentParser(description="Run dev proxy client.")
     parser.add_argument("--relay-url", required=True, help="Relay websocket URL.")
+    parser.add_argument(
+        "--token",
+        help="Shared token for the relay (appended to --relay-url as ?token=...).",
+    )
+    parser.add_argument(
+        "--token-env",
+        default=DEFAULT_TOKEN_ENV_VAR,
+        help=f"Env var name to read token from if --token is unset (default: {DEFAULT_TOKEN_ENV_VAR}).",
+    )
     parser.add_argument(
         "--target-base-url",
         required=True,
@@ -160,11 +195,16 @@ def main(argv: list[str] | None = None) -> None:  # pragma: no cover
     parser = _build_parser()
     args = parser.parse_args(argv)
     override_paths = _parse_override_paths(args.override_path, args.override_paths)
+    try:
+        token = _resolve_token(args.token, args.token_env)
+    except ValueError as exc:
+        parser.error(str(exc))
+    relay_url = _apply_token(args.relay_url, token)
 
     logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
     asyncio.run(
         run_client(
-            relay_url=args.relay_url,
+            relay_url=relay_url,
             target_base_url=args.target_base_url,
             override_paths=override_paths,
             timeout=args.timeout,

src/fastapi_dev_proxy/server.py

@@ -8,7 +8,7 @@
 from typing import cast
 from uuid import uuid4
 
-from fastapi import FastAPI, Request, Response, WebSocket, status
+from fastapi import FastAPI, HTTPException, Request, Response, WebSocket, status
 from fastapi.websockets import WebSocketDisconnect
 from starlette.types import ASGIApp, Receive, Scope, Send
 
@@ -36,19 +36,27 @@ def __init__(
         app: ASGIApp,
         *,
         relay: RelayManager,
-        websocket_path: str = "/fastapi-dev-proxy",
+        path_prefix: str = "/fastapi-dev-proxy",
     ) -> None:
         self._app = app
         self._relay = relay
-        self._websocket_path = normalize_path(websocket_path.rstrip("/") or "/")
+        prefix = normalize_path(path_prefix.rstrip("/") or "/")
+        if prefix == "/":
+            self._reserved_paths = {"/websocket", "/enable", "/disable"}
+        else:
+            self._reserved_paths = {
+                f"{prefix}/websocket",
+                f"{prefix}/enable",
+                f"{prefix}/disable",
+            }
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         if scope.get("type") != "http":
             await self._app(scope, receive, send)
             return
 
         path = scope.get("path") or "/"
-        if normalize_path(path) == self._websocket_path:
+        if normalize_path(path) in self._reserved_paths:
             await self._app(scope, receive, send)
             return
 
@@ -67,11 +75,13 @@ def __init__(
         self,
         *,
         app: FastAPI | None = None,
-        websocket_path: str = "/fastapi-dev-proxy",
-        token: str | None = None,
+        path_prefix: str = "/fastapi-dev-proxy",
+        token: str,
         enabled: bool = True,
         timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
     ) -> None:
+        if not token:
+            raise ValueError("RelayManager token is required (non-empty).")
         self._token = token
         self._enabled = enabled
         self._timeout_seconds = timeout_seconds
@@ -80,7 +90,7 @@ def __init__(
         self._pending: dict[str, asyncio.Future[WebhookResponse]] = {}
         self._lock = asyncio.Lock()
         if app is not None:
-            self._install(app, websocket_path=websocket_path)
+            self._install(app, path_prefix=path_prefix)
 
     def is_enabled(self) -> bool:
         return self._enabled
@@ -95,45 +105,82 @@ def install(
         self,
         app: FastAPI,
         *,
-        websocket_path: str = "/fastapi-dev-proxy",
+        path_prefix: str = "/fastapi-dev-proxy",
     ) -> None:
-        """Register the dev proxy websocket endpoint and HTTP proxy middleware on an existing instance."""
-        self._install(app, websocket_path=websocket_path)
+        """Register the dev proxy endpoints and HTTP proxy middleware on an existing instance.
 
-    @classmethod
-    def for_app(
-        cls,
-        app: FastAPI,
-        *,
-        websocket_path: str = "/fastapi-dev-proxy",
-        token: str | None = None,
-        enabled: bool = True,
-        timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
-    ) -> RelayManager:
-        """Create a RelayManager and register it on the app (websocket + middleware). Returns the relay."""
-        relay = cls(token=token, enabled=enabled, timeout_seconds=timeout_seconds)
-        relay._install(app, websocket_path=websocket_path)
-        return relay
+        Endpoints:
+
+        - `{path_prefix}/websocket` (websocket) connects the dev client
+        - `{path_prefix}/enable` (HTTP) enables forwarding
+        - `{path_prefix}/disable` (HTTP) disables forwarding
+        """
+        self._install(app, path_prefix=path_prefix)
 
     def _install(
         self,
         app: FastAPI,
         *,
-        websocket_path: str = "/fastapi-dev-proxy",
+        path_prefix: str = "/fastapi-dev-proxy",
     ) -> None:
-        """Register the dev proxy websocket endpoint and HTTP proxy middleware."""
+        """Register the dev proxy endpoints and HTTP proxy middleware."""
+        prefix = normalize_path(path_prefix.rstrip("/") or "/")
+        websocket_path = f"{prefix}/websocket" if prefix != "/" else "/websocket"
+        enable_path = f"{prefix}/enable" if prefix != "/" else "/enable"
+        disable_path = f"{prefix}/disable" if prefix != "/" else "/disable"
+
         self._register_websocket(app, websocket_path)
+        self._register_toggle_routes(app, enable_path=enable_path, disable_path=disable_path)
         app.add_middleware(
             RelayProxyMiddleware,
             relay=self,
-            websocket_path=websocket_path,
+            path_prefix=prefix,
         )
 
     def _register_websocket(self, app: FastAPI, websocket_path: str) -> None:
         @app.websocket(websocket_path)
         async def dev_proxy_ws(ws: WebSocket) -> None:
             await self.handle_connection(ws)
 
+    def _register_toggle_routes(self, app: FastAPI, *, enable_path: str, disable_path: str) -> None:
+        @app.post(enable_path)
+        async def dev_proxy_enable(request: Request) -> dict[str, bool]:
+            if not self._is_token_valid(request.query_params.get("token")):
+                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
+            await self.set_enabled(True)
+            return {"enabled": True}
+
+        @app.post(disable_path)
+        async def dev_proxy_disable(request: Request) -> dict[str, bool]:
+            if not self._is_token_valid(request.query_params.get("token")):
+                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
+            await self.set_enabled(False)
+            return {"enabled": False}
+
+    def _is_token_valid(self, token: str | None) -> bool:
+        return bool(token) and token == self._token
+
+    async def set_enabled(self, enabled: bool) -> None:
+        """Enable/disable forwarding.
+
+        If disabling while a dev client is connected, closes the websocket and clears override paths.
+        """
+        async with self._lock:
+            self._enabled = enabled
+            if enabled:
+                return
+
+            websocket = self._websocket
+            if websocket is None:
+                self._override_paths = set()
+                return
+
+            try:
+                await websocket.close(code=status.WS_1001_GOING_AWAY)
+            except Exception:
+                logger.debug("Failed to close dev proxy websocket on disable")
+            await self._clear_connection(websocket)
+
 
     async def handle_connection(self, websocket: WebSocket) -> None:
         """Handle websocket lifecycle and response messages."""
@@ -142,11 +189,9 @@ async def handle_connection(self, websocket: WebSocket) -> None:
             await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
             return
 
-        if self._token is not None:
-            token = websocket.query_params.get("token")
-            if not token or token != self._token:
-                await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
-                return
+        if not self._is_token_valid(websocket.query_params.get("token")):
+            await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
+            return
 
         await websocket.accept()