diff --git a/fastapi_mcp/server.py b/fastapi_mcp/server.py index f5c4fc6..d4178e1 100644 --- a/fastapi_mcp/server.py +++ b/fastapi_mcp/server.py @@ -113,6 +113,15 @@ def __init__( Optional[AuthConfig], Doc("Configuration for MCP authentication"), ] = None, + headers: Annotated[ + Optional[List[str]], + Doc( + """ + List of HTTP header names to forward from the incoming MCP request into each tool invocation. + Only headers in this allowlist will be forwarded. Defaults to ['authorization']. + """ + ), + ] = None, ): # Validate operation and tag filtering options if include_operations is not None and exclude_operations is not None: @@ -147,6 +156,8 @@ def __init__( timeout=10.0, ) + self._forward_headers = {h.lower() for h in (headers or ["Authorization"])} + self.setup_server() def setup_server(self) -> None: @@ -407,11 +418,12 @@ async def _execute_api_tool( raise ValueError(f"Parameter name is None for parameter: {param}") headers[param_name] = arguments.pop(param_name) + # Forward headers that are in the allowlist if http_request_info and http_request_info.headers: - if "Authorization" in http_request_info.headers: - headers["Authorization"] = http_request_info.headers["Authorization"] - elif "authorization" in http_request_info.headers: - headers["Authorization"] = http_request_info.headers["authorization"] + for name, value in http_request_info.headers.items(): + # case-insensitive check for allowed headers + if name.lower() in self._forward_headers: + headers[name] = value body = arguments if arguments else None diff --git a/tests/test_mcp_simple_app.py b/tests/test_mcp_simple_app.py index 64c14d4..c4ba603 100644 --- a/tests/test_mcp_simple_app.py +++ b/tests/test_mcp_simple_app.py @@ -22,6 +22,18 @@ def fastapi_mcp(simple_fastapi_app: FastAPI) -> FastApiMCP: return mcp +@pytest.fixture +def fastapi_mcp_with_custom_header(simple_fastapi_app: FastAPI) -> FastApiMCP: + mcp = FastApiMCP( + simple_fastapi_app, + name="Test MCP Server with custom header", + description="Test description", + headers=["X-Custom-Header"], + ) + mcp.mount() + return mcp + + @pytest.fixture def lowlevel_server_simple_app(fastapi_mcp: FastApiMCP) -> Server: return fastapi_mcp.server @@ -311,5 +323,48 @@ async def test_headers_passthrough_to_tool_handler(fastapi_mcp: FastApiMCP): if mock_request.called: headers_arg = mock_request.call_args[0][4] # headers are the 5th argument - assert "Authorization" in headers_arg - assert headers_arg["Authorization"] == "Bearer token456" + assert "authorization" in headers_arg + assert headers_arg["authorization"] == "Bearer token456" + + +@pytest.mark.asyncio +async def test_custom_header_passthrough_to_tool_handler(fastapi_mcp_with_custom_header: FastApiMCP): + from unittest.mock import patch, MagicMock + from fastapi_mcp.types import HTTPRequestInfo + + # Test with custom header "X-Custom-Header" + with patch.object(fastapi_mcp_with_custom_header, "_request") as mock_request: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = '{"result": "success"}' + mock_response.json.return_value = {"result": "success"} + mock_request.return_value = mock_response + + http_request_info = HTTPRequestInfo( + method="POST", + path="/test", + headers={"X-Custom-Header": "MyValue123"}, + cookies={}, + query_params={}, + body=None, + ) + + try: + # Call the _execute_api_tool method directly + # We don't care if it succeeds, just that _request gets the right headers + await fastapi_mcp_with_custom_header._execute_api_tool( + client=fastapi_mcp_with_custom_header._http_client, + tool_name="get_item", + arguments={"item_id": 1}, + operation_map=fastapi_mcp_with_custom_header.operation_map, + http_request_info=http_request_info, + ) + except Exception: + pass + + assert mock_request.called, "The _request method was not called" + + if mock_request.called: + headers_arg = mock_request.call_args[0][4] # headers are the 5th argument + assert "X-Custom-Header" in headers_arg + assert headers_arg["X-Custom-Header"] == "MyValue123"