feat: add per-request HTTP headers support to ClientSession

Implements per-request headers functionality across all
ClientSession methods to enable multi-tenant authentication, request
tracing, A/B testing, and debugging while maintaining a single
persistent connection.

Technical Implementation:
- Headers are passed through ClientMessageMetadata to transport layer
- Per-request headers take precedence over connection-level headers
- All methods follow consistent extra_headers: dict[str, str] | None = None pattern

Addresses: #1509

Author: damianoneill

README.md

@@ -2267,6 +2267,109 @@ 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 headers functionality for MCP client.
+
+Shows how to use the extra_headers parameter to send different HTTP headers
+with each request, enabling use cases like per-user authentication, request
+tracing, A/B testing, 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"}
+
+    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: Request tracing
+            tracing_headers = {
+                "X-Request-ID": "req-12345",
+                "X-Trace-ID": "trace-abc-456",
+            }
+            result = await session.call_tool("process_data", {"type": "analytics"}, extra_headers=tracing_headers)
+            print(f"Traced request result: {result}")
+
+            # Example 2: User-specific authentication
+            user_headers = {
+                "X-User-ID": "alice",
+                "X-Auth-Token": "user-token-12345",
+            }
+            result = await session.call_tool("get_user_data", {"fields": ["profile"]}, extra_headers=user_headers)
+            print(f"User-specific result: {result}")
+
+            # Example 3: A/B testing
+            experiment_headers = {
+                "X-Experiment-ID": "new-ui-test",
+                "X-Variant": "variant-b",
+            }
+            result = await session.call_tool(
+                "get_recommendations", {"user_id": "user123"}, extra_headers=experiment_headers
+            )
+            print(f"A/B test result: {result}")
+
+            # Example 4: Override connection-level headers
+            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"Admin operation result: {result}")
+
+            # Example 5: Works with all ClientSession methods
+            await session.list_resources(extra_headers={"X-Resource-Filter": "public"})
+            await session.get_prompt("template", extra_headers={"X-Context": "help"})
+            await session.set_logging_level("debug", extra_headers={"X-Debug-Session": "true"})
+
+
+if __name__ == "__main__":
+    print("MCP Client Per-Request Headers Example")
+    print("=" * 50)
+
+    try:
+        asyncio.run(main())
+    except Exception as e:
+        print(f"Example requires a running MCP server. Error: {e}")
+        print("\nThis example demonstrates the API usage patterns.")
+```
+
+_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:

examples/snippets/clients/per_request_headers_example.py

@@ -0,0 +1,77 @@
+"""
+Example demonstrating per-request headers functionality for MCP client.
+
+Shows how to use the extra_headers parameter to send different HTTP headers
+with each request, enabling use cases like per-user authentication, request
+tracing, A/B testing, 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"}
+
+    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: Request tracing
+            tracing_headers = {
+                "X-Request-ID": "req-12345",
+                "X-Trace-ID": "trace-abc-456",
+            }
+            result = await session.call_tool("process_data", {"type": "analytics"}, extra_headers=tracing_headers)
+            print(f"Traced request result: {result}")
+
+            # Example 2: User-specific authentication
+            user_headers = {
+                "X-User-ID": "alice",
+                "X-Auth-Token": "user-token-12345",
+            }
+            result = await session.call_tool("get_user_data", {"fields": ["profile"]}, extra_headers=user_headers)
+            print(f"User-specific result: {result}")
+
+            # Example 3: A/B testing
+            experiment_headers = {
+                "X-Experiment-ID": "new-ui-test",
+                "X-Variant": "variant-b",
+            }
+            result = await session.call_tool(
+                "get_recommendations", {"user_id": "user123"}, extra_headers=experiment_headers
+            )
+            print(f"A/B test result: {result}")
+
+            # Example 4: Override connection-level headers
+            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"Admin operation result: {result}")
+
+            # Example 5: Works with all ClientSession methods
+            await session.list_resources(extra_headers={"X-Resource-Filter": "public"})
+            await session.get_prompt("template", extra_headers={"X-Context": "help"})
+            await session.set_logging_level("debug", extra_headers={"X-Debug-Session": "true"})
+
+
+if __name__ == "__main__":
+    print("MCP Client Per-Request Headers Example")
+    print("=" * 50)
+
+    try:
+        asyncio.run(main())
+    except Exception as e:
+        print(f"Example requires a running MCP server. Error: {e}")
+        print("\nThis example demonstrates the API usage patterns.")