Add configurable HTTP header forwarding

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
 

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"