mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-10-16 01:43:20 +02:00
a34677e10cBackports #4843 from main. `get_http_headers()` now treats `cookie` as a credential header alongside `authorization`, so it is left out of the default mapping. The OpenAPI provider and the proxy transports build outgoing requests from that mapping, so their upstreams receive only the cookies the server configures itself: cookies or headers set on the OpenAPI `httpx` client, and cookie parameters the operation declares. Proxy transports continue to forward `Authorization`. `CurrentHeaders()` still includes both credential headers, since it exposes the current request to the handler. Code that calls `get_http_headers()` and needs the cookie can ask for it by name: ```python from fastmcp.server.dependencies import get_http_headers headers = get_http_headers(include={"cookie"}) ``` Co-authored-by: Claude <noreply@anthropic.com>
310 lines
11 KiB
Python
310 lines
11 KiB
Python
import json
|
|
from typing import Literal
|
|
|
|
import pytest
|
|
from mcp.types import TextContent, TextResourceContents
|
|
from starlette.requests import Request
|
|
|
|
from fastmcp.client import Client
|
|
from fastmcp.client.transports import (
|
|
ClientTransport,
|
|
SSETransport,
|
|
StreamableHttpTransport,
|
|
)
|
|
from fastmcp.server.dependencies import (
|
|
CurrentHeaders,
|
|
CurrentRequest,
|
|
get_http_headers,
|
|
get_http_request,
|
|
)
|
|
from fastmcp.server.server import FastMCP, create_proxy
|
|
from fastmcp.utilities.tests import run_server_async
|
|
|
|
|
|
def fastmcp_server():
|
|
server = FastMCP()
|
|
|
|
# Add a tool
|
|
@server.tool
|
|
def get_headers_tool() -> dict[str, str]:
|
|
"""Get the HTTP headers from the request."""
|
|
request = get_http_request()
|
|
|
|
return dict(request.headers)
|
|
|
|
@server.resource(uri="request://headers")
|
|
async def get_headers_resource() -> str:
|
|
import json
|
|
|
|
request = get_http_request()
|
|
return json.dumps(dict(request.headers))
|
|
|
|
# Add a prompt
|
|
@server.prompt
|
|
def get_headers_prompt() -> str:
|
|
"""Get the HTTP headers from the request."""
|
|
request = get_http_request()
|
|
|
|
return json.dumps(dict(request.headers))
|
|
|
|
return server
|
|
|
|
|
|
@pytest.fixture
|
|
async def shttp_server():
|
|
"""Start a test server with StreamableHttp transport."""
|
|
server = fastmcp_server()
|
|
async with run_server_async(server, transport="http") as url:
|
|
yield url
|
|
|
|
|
|
@pytest.fixture
|
|
async def sse_server():
|
|
"""Start a test server with SSE transport."""
|
|
server = fastmcp_server()
|
|
async with run_server_async(server, transport="sse") as url:
|
|
yield url
|
|
|
|
|
|
async def test_http_headers_resource_shttp(shttp_server: str):
|
|
"""Test getting HTTP headers from the server."""
|
|
async with Client(
|
|
transport=StreamableHttpTransport(
|
|
shttp_server, headers={"X-DEMO-HEADER": "ABC"}
|
|
)
|
|
) as client:
|
|
raw_result = await client.read_resource("request://headers")
|
|
assert isinstance(raw_result[0], TextResourceContents)
|
|
json_result = json.loads(raw_result[0].text)
|
|
assert "x-demo-header" in json_result
|
|
assert json_result["x-demo-header"] == "ABC"
|
|
|
|
|
|
async def test_http_headers_resource_sse(sse_server: str):
|
|
"""Test getting HTTP headers from the server."""
|
|
async with Client(
|
|
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
|
|
) as client:
|
|
raw_result = await client.read_resource("request://headers")
|
|
assert isinstance(raw_result[0], TextResourceContents)
|
|
json_result = json.loads(raw_result[0].text)
|
|
assert "x-demo-header" in json_result
|
|
assert json_result["x-demo-header"] == "ABC"
|
|
|
|
|
|
async def test_http_headers_tool_shttp(shttp_server: str):
|
|
"""Test getting HTTP headers from the server."""
|
|
async with Client(
|
|
transport=StreamableHttpTransport(
|
|
shttp_server, headers={"X-DEMO-HEADER": "ABC"}
|
|
)
|
|
) as client:
|
|
result = await client.call_tool("get_headers_tool")
|
|
assert "x-demo-header" in result.data
|
|
assert result.data["x-demo-header"] == "ABC"
|
|
|
|
|
|
async def test_http_headers_tool_sse(sse_server: str):
|
|
async with Client(
|
|
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
|
|
) as client:
|
|
result = await client.call_tool("get_headers_tool")
|
|
assert "x-demo-header" in result.data
|
|
assert result.data["x-demo-header"] == "ABC"
|
|
|
|
|
|
async def test_http_headers_prompt_shttp(shttp_server: str):
|
|
"""Test getting HTTP headers from the server."""
|
|
async with Client(
|
|
transport=StreamableHttpTransport(
|
|
shttp_server, headers={"X-DEMO-HEADER": "ABC"}
|
|
)
|
|
) as client:
|
|
result = await client.get_prompt("get_headers_prompt")
|
|
assert isinstance(result.messages[0].content, TextContent)
|
|
json_result = json.loads(result.messages[0].content.text)
|
|
assert "x-demo-header" in json_result
|
|
assert json_result["x-demo-header"] == "ABC"
|
|
|
|
|
|
async def test_http_headers_prompt_sse(sse_server: str):
|
|
"""Test getting HTTP headers from the server."""
|
|
async with Client(
|
|
transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"})
|
|
) as client:
|
|
result = await client.get_prompt("get_headers_prompt")
|
|
assert isinstance(result.messages[0].content, TextContent)
|
|
json_result = json.loads(result.messages[0].content.text)
|
|
assert "x-demo-header" in json_result
|
|
assert json_result["x-demo-header"] == "ABC"
|
|
|
|
|
|
async def test_get_http_headers_excludes_content_type(sse_server: str):
|
|
"""Test that get_http_headers() excludes content-type header (issue #3097).
|
|
|
|
This prevents HTTP 415 errors when forwarding headers to downstream APIs
|
|
that require specific Content-Type headers (e.g., application/vnd.api+json).
|
|
"""
|
|
from fastmcp.server.dependencies import get_http_headers
|
|
|
|
server = FastMCP()
|
|
|
|
@server.tool
|
|
def check_excluded_headers() -> dict[str, str]:
|
|
"""Check that problematic headers are excluded from get_http_headers()."""
|
|
return get_http_headers()
|
|
|
|
async with run_server_async(server, transport="sse") as url:
|
|
async with Client(
|
|
transport=SSETransport(
|
|
url,
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
"X-Custom-Header": "should-be-included",
|
|
},
|
|
)
|
|
) as client:
|
|
result = await client.call_tool("check_excluded_headers")
|
|
headers = result.data
|
|
|
|
# These headers should be excluded
|
|
assert "content-type" not in headers
|
|
assert "accept" not in headers
|
|
assert "host" not in headers
|
|
assert "content-length" not in headers
|
|
|
|
# Custom headers should be included
|
|
assert "x-custom-header" in headers
|
|
assert headers["x-custom-header"] == "should-be-included"
|
|
|
|
|
|
async def test_get_http_headers_excludes_cookie():
|
|
"""get_http_headers() withholds Cookie unless the caller asks for it."""
|
|
server = FastMCP()
|
|
|
|
@server.tool
|
|
def default_headers() -> dict[str, str]:
|
|
return get_http_headers()
|
|
|
|
@server.tool
|
|
def opted_in_headers() -> dict[str, str]:
|
|
return get_http_headers(include={"cookie"})
|
|
|
|
async with run_server_async(server, transport="sse") as url:
|
|
async with Client(
|
|
transport=SSETransport(
|
|
url,
|
|
headers={"Cookie": "session=alice-marker", "X-Keep": "yes"},
|
|
)
|
|
) as client:
|
|
default = (await client.call_tool("default_headers")).data
|
|
assert "cookie" not in default
|
|
assert default["x-keep"] == "yes"
|
|
|
|
opted_in = (await client.call_tool("opted_in_headers")).data
|
|
assert opted_in["cookie"] == "session=alice-marker"
|
|
|
|
|
|
async def test_current_headers_exposes_cookie():
|
|
"""CurrentHeaders() reads the current request, so credentials stay visible."""
|
|
server = FastMCP()
|
|
|
|
@server.tool
|
|
def read_request(headers: dict = CurrentHeaders()) -> dict[str, str]:
|
|
return headers
|
|
|
|
async with run_server_async(server, transport="sse") as url:
|
|
async with Client(
|
|
transport=SSETransport(
|
|
url,
|
|
headers={
|
|
"Cookie": "session=alice-marker",
|
|
"Authorization": "Bearer alice-token",
|
|
},
|
|
)
|
|
) as client:
|
|
headers = (await client.call_tool("read_request")).data
|
|
assert headers["cookie"] == "session=alice-marker"
|
|
assert headers["authorization"] == "Bearer alice-token"
|
|
|
|
|
|
@pytest.mark.parametrize("backend_transport", ["http", "sse"])
|
|
async def test_proxy_forwards_authorization_but_not_cookie(
|
|
backend_transport: Literal["http", "sse"],
|
|
):
|
|
"""Proxy transports forward the caller's authorization, not its cookies."""
|
|
backend = fastmcp_server()
|
|
async with run_server_async(backend, transport=backend_transport) as backend_url:
|
|
upstream: ClientTransport
|
|
if backend_transport == "sse":
|
|
upstream = SSETransport(backend_url)
|
|
else:
|
|
upstream = StreamableHttpTransport(backend_url)
|
|
proxy = create_proxy(upstream)
|
|
async with run_server_async(proxy, transport="http") as proxy_url:
|
|
async with Client(
|
|
transport=StreamableHttpTransport(
|
|
proxy_url,
|
|
headers={
|
|
"Cookie": "session=alice-marker",
|
|
"Authorization": "Bearer alice-token",
|
|
"X-Keep": "yes",
|
|
},
|
|
)
|
|
) as client:
|
|
result = await client.call_tool("get_headers_tool")
|
|
|
|
assert "cookie" not in result.data
|
|
assert result.data["authorization"] == "Bearer alice-token"
|
|
assert result.data["x-keep"] == "yes"
|
|
|
|
|
|
async def test_background_task_can_read_snapshotted_request_headers():
|
|
"""Background tools can still access request headers via get_http_request()."""
|
|
server = FastMCP()
|
|
|
|
@server.tool(task=True)
|
|
async def check_request_header() -> str:
|
|
request = get_http_request()
|
|
return request.headers.get("x-tenant-id", "missing")
|
|
|
|
async with run_server_async(server, transport="sse") as url:
|
|
async with Client(
|
|
transport=SSETransport(url, headers={"X-Tenant-ID": "tenant-123"})
|
|
) as client:
|
|
task = await client.call_tool("check_request_header", task=True)
|
|
result = await task.result()
|
|
assert result.data == "tenant-123"
|
|
|
|
|
|
async def test_background_task_current_http_dependencies_restore_headers():
|
|
"""CurrentHeaders/CurrentRequest work in task workers without explicit Context."""
|
|
server = FastMCP()
|
|
|
|
@server.tool(task=True)
|
|
async def check_headers(
|
|
headers: dict[str, str] = CurrentHeaders(),
|
|
request: Request = CurrentRequest(),
|
|
) -> dict[str, str]:
|
|
return {
|
|
"authorization": headers.get("authorization", "missing"),
|
|
"tenant": request.headers.get("x-tenant-id", "missing"),
|
|
}
|
|
|
|
async with run_server_async(server, transport="sse") as url:
|
|
async with Client(
|
|
transport=SSETransport(
|
|
url,
|
|
headers={
|
|
"Authorization": "Bearer tenant-token",
|
|
"X-Tenant-ID": "tenant-456",
|
|
},
|
|
)
|
|
) as client:
|
|
task = await client.call_tool("check_headers", task=True)
|
|
result = await task.result()
|
|
assert result.data == {
|
|
"authorization": "Bearer tenant-token",
|
|
"tenant": "tenant-456",
|
|
}
|