Support routable transport headers for gateways (SEP-2243)

docs/deployment/http.mdx

@@ -149,6 +149,42 @@ export FASTMCP_HTTP_ALLOWED_ORIGINS='["https://app.example.com"]'
 
 Use `host_origin_protection="auto"` to protect localhost-bound direct servers while allowing ASGI, serverless, and reverse-proxy deployments to keep their existing Host handling unless they configure explicit trust rules. Use `host_origin_protection=False` to keep the request guard disabled.
 
+### Gateway Routing Headers
+
+A gateway, load balancer, or reverse proxy in front of your MCP server often needs to route a request before it reads the JSON-RPC body — the body may be an SSE stream, or the gateway may simply want to avoid parsing it. Streamable HTTP clients built on the MCP Python SDK (including FastMCP's own client) attach routing information to each request as HTTP headers so an intermediary can dispatch on headers alone:
+
+- `Mcp-Method` carries the JSON-RPC method (for example `tools/call`) on every request.
+- `Mcp-Name` carries the target's name on named operations — the tool name for `tools/call`, the prompt name for `prompts/get`, the resource URI for `resources/read`.
+- `Mcp-Param-*` carries selected argument values for a `tools/call`, one header per opted-in parameter.
+
+FastMCP's HTTP transport neither strips nor rewrites these headers, so a gateway sees them exactly as the client sent them. The `Host`/`Origin` request guard inspects only `Host` and `Origin` and leaves the routing headers untouched.
+
+To expose an argument as an `Mcp-Param-*` header, annotate the parameter with the `x-mcp-header` JSON Schema extension. FastMCP carries the annotation into the tool's advertised input schema, and a conforming client mirrors the argument into a header named `Mcp-Param-<token>`:
+
+```python
+from typing import Annotated
+
+from pydantic import Field
+
+from fastmcp import FastMCP
+
+mcp = FastMCP("My Server")
+
+@mcp.tool
+def query_tenant(
+    tenant: Annotated[str, Field(json_schema_extra={"x-mcp-header": "Tenant"})],
+    sql: str,
+) -> str:
+    """A call to this tool sends the tenant value as an `Mcp-Param-Tenant` header."""
+    ...
+```
+
+A gateway can now route on `Mcp-Param-Tenant` — for example, pinning each tenant to a dedicated backend — without inspecting the request body. The annotation is only permitted on `string`, `integer`, and `boolean` parameters. These headers advertise routing intent; treat them as untrusted hints, since the server still validates the request body as the source of truth.
+
+<Tip>
+When you put a FastMCP [proxy](/servers/proxy) in front of another server, the proxy re-advertises each backend tool's `x-mcp-header` annotation, so routing headers work across the proxy hop as well. The headers themselves are regenerated per hop rather than forwarded verbatim, since each describes a single HTTP request.
+</Tip>
+
 ### Health Checks
 
 Health check endpoints are essential for monitoring your deployed server and ensuring it's responding correctly. FastMCP allows you to add custom routes alongside your MCP endpoints, making it easy to implement health checks that work with both deployment approaches.

tests/server/http/test_routable_headers.py

@@ -0,0 +1,89 @@
+"""Routable transport headers (SEP-2243) survive a FastMCP HTTP round trip.
+
+The MCP Python SDK emits the routing headers on the client (`ClientSession`) and
+validates them on the modern streamable-HTTP server transport. These tests are
+FastMCP's regression guard: they prove FastMCP's HTTP layer neither strips nor
+blocks the headers, so a gateway sitting in front of a FastMCP server can route
+on them. The tool echoes back the raw request headers it received, letting the
+test assert on exactly what reached the server.
+"""
+
+from typing import Annotated
+
+from pydantic import Field
+
+from fastmcp.server.dependencies import get_http_request
+from fastmcp.server.server import FastMCP
+from fastmcp.utilities.tests import asgi_server
+
+
+def _echo_server() -> FastMCP:
+    server = FastMCP()
+
+    @server.tool
+    def echo_headers(
+        tenant: Annotated[
+            str, Field(json_schema_extra={"x-mcp-header": "Tenant"})
+        ] = "acme",
+    ) -> dict[str, str]:
+        """Return the raw HTTP headers the server received for this request."""
+        return dict(get_http_request().headers)
+
+    return server
+
+
+async def test_mcp_method_and_name_headers_reach_server():
+    """`Mcp-Method` and `Mcp-Name` set by the SDK client arrive at the server."""
+    async with asgi_server(_echo_server(), transport="http") as running_server:
+        async with running_server.client() as client:
+            result = await client.call_tool("echo_headers")
+
+    headers = result.data
+    assert headers["mcp-method"] == "tools/call"
+    assert headers["mcp-name"] == "echo_headers"
+
+
+async def test_mcp_param_header_reaches_server():
+    """An `x-mcp-header` annotated parameter is mirrored into `Mcp-Param-*`.
+
+    The SDK client only emits `Mcp-Param-*` once it has seen the tool's input
+    schema, so the test lists tools before calling.
+    """
+    async with asgi_server(_echo_server(), transport="http") as running_server:
+        async with running_server.client() as client:
+            await client.list_tools()
+            result = await client.call_tool("echo_headers", {"tenant": "beta-corp"})
+
+    headers = result.data
+    assert headers["mcp-param-tenant"] == "beta-corp"
+
+
+async def test_routing_headers_survive_host_origin_protection():
+    """The Host/Origin request guard does not strip the routing headers."""
+    async with asgi_server(
+        _echo_server(),
+        transport="http",
+        host_origin_protection=True,
+        allowed_hosts=["*"],
+        allowed_origins=["*"],
+    ) as running_server:
+        async with running_server.client() as client:
+            await client.list_tools()
+            result = await client.call_tool("echo_headers", {"tenant": "gamma"})
+
+    headers = result.data
+    assert headers["mcp-method"] == "tools/call"
+    assert headers["mcp-name"] == "echo_headers"
+    assert headers["mcp-param-tenant"] == "gamma"
+
+
+async def test_x_mcp_header_annotation_survives_schema_generation():
+    """FastMCP preserves `x-mcp-header` in a tool's advertised input schema.
+
+    This is the annotation the SDK client reads to decide which arguments to
+    mirror into `Mcp-Param-*` headers, so it must reach the wire unchanged.
+    """
+    server = _echo_server()
+    tools = await server._list_tools()
+    (tool,) = [t for t in tools if t.name == "echo_headers"]
+    assert tool.parameters["properties"]["tenant"]["x-mcp-header"] == "Tenant"

tests/server/providers/proxy/test_proxy_server.py

@@ -1554,3 +1554,31 @@ async def test_connection_error_reaches_client_on_both_eras(self, mode: str):
         with pytest.raises(MCPError, match="Client failed to connect"):
             async with Client(proxy, mode=mode) as client:
                 await client.list_tools()
+
+
+async def test_proxy_preserves_x_mcp_header_annotation():
+    """A proxy re-advertises a backend tool's `x-mcp-header` annotation (SEP-2243).
+
+    The routing headers are per-hop: the SDK client regenerates them on each
+    HTTP request. For `Mcp-Param-*` to be emitted on the proxy->backend hop (and
+    on the caller->proxy hop), the proxy must carry the backend's `x-mcp-header`
+    schema annotation through to its own advertised tool schema.
+    """
+    from typing import Annotated
+
+    from pydantic import Field
+
+    backend = FastMCP("Backend")
+
+    @backend.tool
+    def route(
+        tenant: Annotated[str, Field(json_schema_extra={"x-mcp-header": "Tenant"})],
+    ) -> str:
+        return tenant
+
+    proxy = create_proxy(backend)
+    async with Client(proxy) as client:
+        tools = await client.list_tools()
+
+    (tool,) = [t for t in tools if t.name == "route"]
+    assert tool.input_schema["properties"]["tenant"]["x-mcp-header"] == "Tenant"