mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
"""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"
|