feat: add per-request HTTP headers support for call_tool()

damianoneill · damianoneill · commit 40eae958e672 · 2025-10-26T10:19:22.000Z
Implements per-request headers functionality to enable various use cases such as multi-tenant authentication, request tracing, A/B testing, and debugging while maintaining a single persistent connection. Changes: - Add extra_headers parameter to ClientSession.call_tool() - Extend ClientMessageMetadata to support extra_headers - Update StreamableHTTPTransport to merge per-request headers - Add tests - Include usage examples and documentation - Maintain full backward compatibility This addresses GitHub issues #1509 and supports multi-tenant scenarios where different requests require different authentication tokens or contextual headers. Future work will extend extra_headers support to other ClientSession methods (get_prompt, read_resource, etc.) based on maintainer feedback.
diff --git a/README.md b/README.md
@@ -2153,6 +2153,114 @@ if __name__ == "__main__":
 _Full example: [examples/snippets/clients/streamable_basic.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/streamable_basic.py)_
 <!-- /snippet-source -->
 
+### Per-Request HTTP Headers
+
+When using HTTP transports, you can pass custom headers on a per-request basis. This enables various use cases such as request tracing, authentication context, A/B testing, debugging flags, and more while maintaining a single persistent connection:
+
+<!-- snippet-source examples/snippets/clients/per_request_headers_example.py -->
+```python
+"""
+Example demonstrating per-request HTTP headers functionality.
+Run from the repository root:
+    uv run examples/snippets/clients/per_request_headers_example.py
+"""
+
+import asyncio
+
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+
+
+async def main():
+    """Demonstrate per-request headers usage."""
+    # Connect to a streamable HTTP server
+    async with streamablehttp_client("http://localhost:8000/mcp") as (
+        read_stream,
+        write_stream,
+        _,
+    ):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+
+            # Example 1: Request tracing and debugging
+            result = await session.call_tool(
+                "analyze_data",
+                arguments={"dataset": "sales_q4"},
+                extra_headers={
+                    "X-Request-ID": "req-12345",
+                    "X-Debug-Mode": "true",
+                    "X-Trace-ID": "trace-abc-456"
+                }
+            )
+
+            print(f"Result with tracing: {result}")
+
+            # Example 2: A/B testing context
+            result = await session.call_tool(
+                "get_recommendations",
+                arguments={"user_id": "user123"},
+                extra_headers={
+                    "X-Experiment-ID": "rec-algo-v2",
+                    "X-Variant": "variant-b"
+                }
+            )
+
+            print(f"Result with A/B test context: {result}")
+
+            # Example 3: Per-user authentication context
+            users = [
+                {"id": "user1", "token": "token_abc123"},
+                {"id": "user2", "token": "token_def456"},
+                {"id": "user3", "token": "token_ghi789"},
+            ]
+
+            for user in users:
+                print(f"Making request for {user['id']}")
+
+                # Call tool with user-specific authentication header
+                result = await session.call_tool(
+                    "get_user_data",
+                    arguments={"user_id": user["id"]},
+                    extra_headers={"Authorization": f"Bearer {user['token']}"}
+                )
+
+                print(f"Result for {user['id']}: {result}")
+
+            # Headers can also be combined with other per-request parameters
+            result = await session.call_tool(
+                "slow_operation",
+                arguments={"data": "example"},
+                extra_headers={
+                    "X-Request-ID": "req-12345",
+                    "X-Priority": "high"
+                },
+                read_timeout_seconds=30.0  # Extended timeout for slow operation
+            )
+
+            print(f"Slow operation result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+```
+
+_Full example: [examples/snippets/clients/per_request_headers_example.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/per_request_headers_example.py)_
+<!-- /snippet-source -->
+
+The `extra_headers` parameter is available for all `ClientSession` methods that make server requests:
+- `call_tool()`
+- `get_prompt()`
+- `read_resource()`
+- `list_tools()`
+- `list_prompts()`
+- `list_resources()`
+- `list_resource_templates()`
+- `subscribe()`
+- `unsubscribe()`
+- `set_logging_level()`
+
+Per-request headers are merged with the transport's default headers, with per-request headers taking precedence for duplicate keys.
+
 ### Client Display Utilities
 
 When building MCP clients, the SDK provides utilities to help display human-readable names for tools, resources, and prompts:
diff --git a/examples/snippets/clients/per_request_headers_example.py b/examples/snippets/clients/per_request_headers_example.py
@@ -0,0 +1,162 @@
+"""
+Example demonstrating per-request headers functionality for MCP client.
+
+This example shows how to use the new extra_headers parameter in call_tool()
+to send different HTTP headers with each tool call, enabling various use cases
+such as per-user authentication, request tracing, A/B testing, debugging flags,
+and multi-tenant applications.
+"""
+
+import asyncio
+
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+
+
+async def main():
+    """Demonstrate per-request headers functionality."""
+
+    # Connection-level headers (static for the entire session)
+    connection_headers = {"Authorization": "Bearer org-level-token", "X-Org-ID": "org-123"}
+
+    # Connect to MCP server with connection-level headers
+    async with streamablehttp_client("https://mcp.example.com/mcp", headers=connection_headers) as (
+        read_stream,
+        write_stream,
+        _,
+    ):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+
+            # Example 1: Call tool without per-request headers
+            # Uses only connection-level headers
+            print("=== Example 1: Default headers ===")
+            result = await session.call_tool("get_data", {})
+            print(f"Result: {result}")
+
+            # Example 2: Request tracing and correlation
+            print("\n=== Example 2: Request tracing ===")
+            tracing_headers = {
+                "X-Request-ID": "req-12345",
+                "X-Trace-ID": "trace-abc-456",
+                "X-Correlation-ID": "corr-789",
+            }
+
+            result = await session.call_tool("process_data", {"type": "analytics"}, extra_headers=tracing_headers)
+            print(f"Result with tracing: {result}")
+
+            # Example 3: A/B testing and feature flags
+            print("\n=== Example 3: A/B testing ===")
+            experiment_headers = {
+                "X-Experiment-ID": "new-ui-test",
+                "X-Variant": "variant-b",
+                "X-Feature-Flags": "enable-beta-features,new-algorithm",
+            }
+
+            result = await session.call_tool(
+                "get_recommendations", {"user_id": "user123"}, extra_headers=experiment_headers
+            )
+            print(f"Result with A/B testing: {result}")
+
+            # Example 4: Debug and profiling
+            print("\n=== Example 4: Debug mode ===")
+            debug_headers = {"X-Debug-Mode": "true", "X-Profile": "performance", "X-Log-Level": "verbose"}
+
+            result = await session.call_tool("complex_calculation", {"data": [1, 2, 3]}, extra_headers=debug_headers)
+            print(f"Result with debugging: {result}")
+
+            # Example 5: Authentication context (user-specific)
+            print("\n=== Example 5: User authentication ===")
+            user_headers = {"X-Auth-Token": "user-token-12345", "X-User-ID": "alice", "X-Session-ID": "sess-789"}
+
+            result = await session.call_tool(
+                "get_user_data", {"fields": ["profile", "preferences"]}, extra_headers=user_headers
+            )
+            print(f"Result for user: {result}")
+
+            # Example 6: Override connection-level headers
+            print("\n=== Example 6: Override connection-level authorization ===")
+            override_headers = {
+                "Authorization": "Bearer user-specific-token",  # Overrides connection-level
+                "X-Special-Permission": "admin",
+            }
+
+            result = await session.call_tool("admin_operation", {"operation": "reset"}, extra_headers=override_headers)
+            print(f"Result with overridden auth: {result}")
+
+            # Example 7: Combine with other call_tool parameters
+            print("\n=== Example 7: Combined with meta and other parameters ===")
+            combined_headers = {"X-Request-Source": "api", "X-Priority": "high", "X-Client-Version": "1.2.3"}
+
+            meta_data = {"correlation_id": "req-123", "client_version": "1.0.0"}
+
+            result = await session.call_tool(
+                "complex_operation",
+                {"param1": "value1", "param2": "value2"},
+                meta=meta_data,
+                extra_headers=combined_headers,
+            )
+            print(f"Result with combined parameters: {result}")
+
+
+# Multi-tenant example showing how different users can use the same connection
+async def multi_tenant_example():
+    """Example of multi-tenant usage with per-request headers."""
+
+    print("\n" + "=" * 60)
+    print("MULTI-TENANT EXAMPLE")
+    print("=" * 60)
+
+    # Organization-level connection
+    org_headers = {"Authorization": "Bearer org-api-key-xyz789", "X-Org-ID": "org-acme-corp"}
+
+    async with streamablehttp_client("https://mcp.example.com/mcp", headers=org_headers) as (
+        read_stream,
+        write_stream,
+        _,
+    ):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+
+            # Simulate handling requests from different tenants/users
+            tenants = [
+                {"tenant_id": "tenant-001", "user_id": "alice", "auth_token": "alice-jwt-token-abc123"},
+                {"tenant_id": "tenant-002", "user_id": "bob", "auth_token": "bob-jwt-token-def456"},
+                {"tenant_id": "tenant-001", "user_id": "charlie", "auth_token": "charlie-jwt-token-ghi789"},
+            ]
+
+            for i, tenant in enumerate(tenants, 1):
+                print(f"\n--- Request {i}: {tenant['user_id']} from {tenant['tenant_id']} ---")
+
+                # Each request gets tenant-specific headers
+                tenant_headers = {
+                    "X-Tenant-ID": tenant["tenant_id"],
+                    "X-User-ID": tenant["user_id"],
+                    "X-Auth-Token": tenant["auth_token"],
+                    "X-Request-ID": f"req-{i}-{tenant['user_id']}",
+                }
+
+                try:
+                    result = await session.call_tool(
+                        "get_tenant_data", {"data_type": "dashboard"}, extra_headers=tenant_headers
+                    )
+                    print(f"Success for {tenant['user_id']}: {len(result.content)} items")
+
+                except Exception as e:
+                    print(f"Error for {tenant['user_id']}: {e}")
+
+
+if __name__ == "__main__":
+    print("MCP Client Per-Request Headers Example")
+    print("=" * 50)
+
+    # Note: This example assumes a running MCP server at the specified URL
+    # In practice, you would replace with your actual MCP server endpoint
+
+    try:
+        asyncio.run(main())
+        asyncio.run(multi_tenant_example())
+    except Exception as e:
+        print(f"Example requires a running MCP server. Error: {e}")
+        print("\nThis example demonstrates the API usage patterns.")
+        print("Replace 'https://mcp.example.com/mcp' with your actual MCP server URL.")
diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py
@@ -330,13 +330,33 @@ async def call_tool(
         progress_callback: ProgressFnT | None = None,
         *,
         meta: dict[str, Any] | None = None,
+        extra_headers: dict[str, str] | None = None,
     ) -> types.CallToolResult:
-        """Send a tools/call request with optional progress callback support."""
+        """Send a tools/call request with optional progress callback support.
+
+        Args:
+            name: The name of the tool to call.
+            arguments: The arguments to pass to the tool.
+            read_timeout_seconds: Optional timeout for reading the response.
+            progress_callback: Optional callback for progress notifications.
+            meta: Optional meta parameters for the request.
+            extra_headers: Additional HTTP headers to include in this specific request.
+                          These are merged with connection-level headers, with extra_headers
+                          taking precedence for duplicate keys. Useful for per-request
+                          authentication, tracing, debugging, A/B testing, and more.
+        """
 
         _meta: types.RequestParams.Meta | None = None
         if meta is not None:
             _meta = types.RequestParams.Meta(**meta)
 
+        # Create metadata to pass extra headers to the transport layer
+        metadata = None
+        if extra_headers:
+            from mcp.shared.message import ClientMessageMetadata
+
+            metadata = ClientMessageMetadata(extra_headers=extra_headers)
+
         result = await self.send_request(
             types.ClientRequest(
                 types.CallToolRequest(
@@ -345,6 +365,7 @@ async def call_tool(
             ),
             types.CallToolResult,
             request_read_timeout_seconds=read_timeout_seconds,
+            metadata=metadata,
             progress_callback=progress_callback,
         )
 
diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py
@@ -220,6 +220,11 @@ async def handle_get_stream(
     async def _handle_resumption_request(self, ctx: RequestContext) -> None:
         """Handle a resumption request using GET with SSE."""
         headers = self._prepare_request_headers(ctx.headers)
+
+        # Merge extra headers from metadata if present
+        if ctx.metadata and ctx.metadata.extra_headers:
+            headers.update(ctx.metadata.extra_headers)
+
         if ctx.metadata and ctx.metadata.resumption_token:
             headers[LAST_EVENT_ID] = ctx.metadata.resumption_token
         else:
@@ -254,6 +259,11 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
     async def _handle_post_request(self, ctx: RequestContext) -> None:
         """Handle a POST request with response processing."""
         headers = self._prepare_request_headers(ctx.headers)
+
+        # Merge extra headers from metadata if present
+        if ctx.metadata and ctx.metadata.extra_headers:
+            headers.update(ctx.metadata.extra_headers)
+
         message = ctx.session_message.message
         is_initialization = self._is_initialization_request(message)
 
diff --git a/src/mcp/shared/message.py b/src/mcp/shared/message.py
@@ -21,6 +21,7 @@ class ClientMessageMetadata:
 
     resumption_token: ResumptionToken | None = None
     on_resumption_token_update: Callable[[ResumptionToken], Awaitable[None]] | None = None
+    extra_headers: dict[str, str] | None = None
 
 
 @dataclass
diff --git a/tests/client/test_extra_headers.py b/tests/client/test_extra_headers.py
