vercel.proxy (draft)

README.md

@@ -71,9 +71,9 @@ asyncio.run(main())
 ## Public API Vs Internals
 
 Import from public modules under `vercel.*`, such as `vercel.blob`,
-`vercel.cache`, `vercel.headers`, `vercel.oidc`, `vercel.projects`, and
-`vercel.sandbox`. Modules under `vercel._internal.*` are implementation details
-and may change without public API guarantees.
+`vercel.cache`, `vercel.headers`, `vercel.oidc`, `vercel.projects`,
+`vercel.proxy`, and `vercel.sandbox`. Modules under `vercel._internal.*` are
+implementation details and may change without public API guarantees.
 
 Sync counterparts are available for the main client classes and module-level
 helpers when you are not running an async application.

changes/vercel-proxy/python-routing-middleware.feature.md

@@ -0,0 +1,2 @@
+Add Starlette-compatible Python routing middleware, route matchers, rewrites,
+redirects, and request continuation helpers.

pyproject.toml

@@ -72,6 +72,7 @@ mypy_path = [
     "src/vercel-headers",
     "src/vercel-queue",
     "src/vercel-oidc",
+    "src/vercel-proxy",
     "src/vercel-internal-telemetry",
     "integrations/vercel-celery",
     "integrations/vercel-dramatiq",

scripts/bundle_release.py

@@ -68,6 +68,10 @@
 PEER_DEPENDENCIES = {
     "vercel-celery": {"celery"},
     "vercel-dramatiq": {"dramatiq"},
+    # Starlette classes are part of vercel.proxy's public interoperability
+    # contract. Vendoring them would make application Response objects fail
+    # identity checks against a second, private Starlette installation.
+    "vercel-proxy": {"starlette"},
 }
 COMMON_DROP_TRANSFORMATIONS = (
     "*.so",

src/vercel-proxy/CHANGELOG.md

@@ -0,0 +1 @@
+# Changelog

src/vercel-proxy/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Vercel, Inc.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

src/vercel-proxy/README.md

@@ -0,0 +1,119 @@
+# Proxy
+
+`vercel.proxy` provides Python routing middleware that runs before Vercel's
+cache and application routing.
+
+## Configuration
+
+Point Vercel at an exported `Proxy` object:
+
+`vercel.json`
+```json
+{
+    "proxy": "proxy.py"
+}
+```
+
+The proxy function has no dependencies installed by default. To add dependencies
+to your proxy, add a `proxy` dependency group in `pyproject.toml`
+
+```toml
+[dependency-groups]
+proxy = ["vercel-proxy"]
+```
+
+
+`proxy.py`:
+
+```python
+from vercel.proxy import Proxy, Request, redirect, rewrite
+
+proxy = Proxy()
+
+
+@proxy.middleware("http")
+async def authenticate(request: Request, call_next):
+    if request.cookies.get("session") is None:
+        return redirect("/login")
+
+    if request.url.path == "/about":
+        return rewrite("/about-2")
+
+    response = await call_next(request)
+    response.headers["x-authenticated"] = "true"
+    return response
+```
+
+`call_next()` advances through the remaining Python proxy middleware and
+routes. It returns a synthetic routing response; it does not invoke, await, or
+contain the eventual CDN or application response.
+
+Returning `None` continues Vercel routing unchanged. Use
+`continue_routing()` when the continuation needs response headers or a
+complete replacement set of request headers:
+
+```python
+from vercel.proxy import continue_routing
+
+return continue_routing(
+    headers={"x-authenticated": "true"},
+    request_headers={
+        **request.headers,
+        "x-user-id": "user_123",
+    },
+)
+```
+
+`request_headers` is the complete set forwarded after the proxy, not a patch.
+
+## Route selected logic
+
+Route paths use template syntax where `{name}` captures a single path segment
+and `{name:path}` captures the remainder of the path. The first route whose
+path, method, and conditions all match is selected. With no `methods`
+argument, a route matches every HTTP method.
+
+```python
+from vercel.proxy import Proxy, Request, Route, redirect, rewrite
+from vercel.proxy.matchers import cookie, header, host, query
+
+
+async def dashboard(request: Request):
+    if request.cookies.get("session") is None:
+        return redirect("/login")
+    return None
+
+
+proxy = Proxy(
+    routes=[
+        Route(
+            "/dashboard/{path:path}",
+            dashboard,
+            has=[host("{tenant}.example.com"), cookie("session")],
+            missing=[header("x-blocked")],
+        ),
+        Route.rewrite(
+            "/legacy/{path:path}",
+            "/new/{path}",
+            has=[query("migrate", "1")],
+        ),
+        Route.redirect("/docs", "/documentation", status_code=308),
+    ]
+)
+```
+
+`header()`, `cookie()`, and `query()` match presence when no value is supplied
+and use exact value matching otherwise. `host()` supports path-style captures;
+captured hostname values are added to `request.path_params`.
+
+For a rewrite that needs arbitrary Python logic, use a normal route handler:
+
+```python
+Route(
+    "/legacy/{path:path}",
+    lambda request: rewrite(f"/new/{request.path_params['path']}"),
+)
+```
+
+Purely static rewrites are generally more efficient in Vercel routing
+configuration because they do not need to start Python.

src/vercel-proxy/hatch_build.py

@@ -0,0 +1,29 @@
+"""Load the shared Vercel Hatch metadata hook."""
+
+from __future__ import annotations
+
+from importlib.util import module_from_spec, spec_from_file_location
+from pathlib import Path
+from types import ModuleType
+
+from hatchling.metadata.plugin.interface import MetadataHookInterface
+
+
+def get_metadata_hook() -> type[MetadataHookInterface]:
+    """Return the shared workspace dependency metadata hook."""
+    return _load_shared_hook().get_metadata_hook()
+
+
+def _load_shared_hook() -> ModuleType:
+    root = Path(__file__).resolve().parent
+    candidates = [root / "../../scripts/hatch_build.py", root / "_vercel_hatch_build.py"]
+    for candidate in candidates:
+        path = candidate.resolve()
+        if path.exists():
+            spec = spec_from_file_location("_vercel_hatch_build", path)
+            if spec is None or spec.loader is None:
+                raise RuntimeError(f"could not load Hatch hook from {path}")
+            module = module_from_spec(spec)
+            spec.loader.exec_module(module)
+            return module
+    raise RuntimeError("could not find shared Vercel Hatch metadata hook")

src/vercel-proxy/pyproject.toml

@@ -0,0 +1,64 @@
+[build-system]
+requires = ["hatchling>=1.27.0,<2"]
+build-backend = "hatchling.build"
+
+[project]
+name = "vercel-proxy"
+dynamic = ["version", "dependencies"]
+description = "Python routing middleware for Vercel"
+readme = "README.md"
+requires-python = ">=3.10"
+license = "MIT"
+license-files = ["LICENSE", "LICENSE.*"]
+
+[tool.hatch.metadata.hooks.custom]
+path = "hatch_build.py"
+
+[tool.vercel.release.dependencies]
+dependencies = [
+    "starlette>=0.46.0,<2",
+]
+
+[tool.hatch.version]
+path = "vercel/proxy/version.py"
+
+[tool.hatch.build.targets.sdist]
+force-include = { "../../scripts/hatch_build.py" = "/_vercel_hatch_build.py" }
+include = [
+    "/vercel/proxy/**/*.py",
+    "/vercel/proxy/py.typed",
+    "/README.md",
+    "/pyproject.toml",
+    "/hatch_build.py",
+    "/LICENSE",
+]
+exclude = [
+    "/**/__pycache__",
+]
+
+[tool.hatch.build.targets.wheel]
+dev-mode-dirs = ["."]
+only-include = [
+    "/vercel/proxy",
+]
+exclude = [
+    "/**/__pycache__",
+]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["."]
+addopts = "--no-header --capture=tee-sys"
+asyncio_mode = "auto"
+
+[tool.poe]
+include = "../../scripts/poe/poe.toml"
+verbosity = -1
+
+[tool.mypy]
+cache_dir = "../../.mypy_cache/vercel-proxy"
+explicit_package_bases = true
+packages = ["vercel.proxy"]
+
+[tool.poe.tasks.typecheck]
+cmd = "$MYPY"

src/vercel-proxy/tests/__init__.py

@@ -0,0 +1 @@
+

src/vercel-proxy/tests/conftest.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+from typing import Any
+
+from starlette.types import ASGIApp, Message
+
+
+def make_scope(
+    path: str = "/",
+    *,
+    method: str = "GET",
+    headers: Mapping[str, str] | Iterable[tuple[str, str]] = (),
+    query_string: str = "",
+) -> dict[str, Any]:
+    header_items = headers.items() if isinstance(headers, Mapping) else headers
+    raw_headers = [(name.lower().encode(), value.encode()) for name, value in header_items]
+    if not any(name == b"host" for name, _ in raw_headers):
+        raw_headers.append((b"host", b"example.com"))
+    return {
+        "type": "http",
+        "asgi": {"version": "3.0", "spec_version": "2.3"},
+        "http_version": "1.1",
+        "method": method,
+        "scheme": "https",
+        "path": path,
+        "raw_path": path.encode(),
+        "query_string": query_string.encode(),
+        "root_path": "",
+        "headers": raw_headers,
+        "client": ("127.0.0.1", 1234),
+        "server": ("example.com", 443),
+    }
+
+
+async def invoke(
+    app: ASGIApp,
+    scope: dict[str, Any] | None = None,
+    *,
+    body: bytes = b"",
+) -> list[Message]:
+    messages: list[Message] = []
+    request_sent = False
+
+    async def receive() -> Message:
+        nonlocal request_sent
+        if request_sent:
+            return {"type": "http.disconnect"}
+        request_sent = True
+        return {"type": "http.request", "body": body, "more_body": False}
+
+    async def send(message: Message) -> None:
+        messages.append(message)
+
+    await app(scope or make_scope(), receive, send)
+    return messages
+
+
+def response_headers(messages: list[Message]) -> dict[str, str]:
+    start = next(message for message in messages if message["type"] == "http.response.start")
+    return {
+        name.decode("latin-1"): value.decode("latin-1") for name, value in start.get("headers", [])
+    }