René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Per-Request Transport Configuration for MCP Clients","articleBody":"## Summary\n\nAdd support for configuring transport settings (e.g., authentication headers, tracing context) dynamically on a per-request basis. This is essential for use cases where settings like authentication tokens or trace IDs need to change between requests.\n\n## Problem Statement\n\nCurrently, MCP clients only supports setting context at initialization time:\n\n1. **HTTP headers** are configured on the `httpx.AsyncClient` at connection establishment\n2. **Session-level context** is fixed at `ClientSession` creation\n3. **Transport-level configuration** is determined when the transport context manager is entered\n\nThis architecture doesn't support common production requirements:\n- **Per-Request Authentication**: Auth tokens that need to be propagated per-request based on the calling user's context\n- **Request Tracing**: Trace IDs/span IDs that change per request\n- **Dynamic Headers**: Headers that vary based on request context (tenant IDs, correlation IDs, etc.)\n\n### Current Workaround Attempts\n\nUsers attempting to propagate context via `contextvars` face challenges:\n- Context vars are copied at session start, not per-request\n- The background thread pattern means context vars set after initialization are not visible\n- Even with context propagation to the background thread, there's no mechanism to pass that context to the HTTP transport layer\n\n### Current API Limitations\n\nThe existing `read_timeout_seconds` parameter on `call_tool()` is a transport-specific concern (HTTP timeout) that was added directly to the session method. This approach doesn't scale as different transports have different configuration needs, and adding transport-specific parameters to session methods leads to API bloat. A generic per-request configuration mechanism would offer more extensibility. \n\nIn addition, this functionality should extend beyond just tool calls, as headers or timeout configurations may be required for all requests to the server. \n\n## Proposed Solution\n\nThere are two levels where context can be provided:\n1. **Per-request** - Pass configuration directly when calling a method\n2. **Session-level provider** - A callback that provides configuration automatically for every request\n\nFor both levels, there are two design approaches for how configuration flows to the transport.\n\n## Design Option A: Transport-Specific Configuration (Recommended)\n\nEach transport exports its own configuration type. Users provide settings in the transport's expected format.\n\n### Complete Example (Option A)\n\n```python\nfrom mcp import ClientSession\nfrom mcp.client.streamable_http import streamable_http_client, HTTPTransportConfig\n\n# === Per-Request Usage ===\nasync def handle_single_request(session: ClientSession, user_token: str):\n    \"\"\"Pass transport config directly on each call.\"\"\"\n    result = await session.call_tool(\n        \"my_tool\",\n        arguments={\"foo\": \"bar\"},\n        # Pass transport config directly - can use typed class or dict\n        transport_config=HTTPTransportConfig(\n            headers={\"Authorization\": f\"Bearer {user_token}\"}\n        )\n        # Or simply: transport_config={\"headers\": {\"Authorization\": f\"Bearer {user_token}\"}}\n    )\n    return result\n\n\n# === Session-Level Provider Usage ===\nasync def my_transport_provider(request: types.ClientRequest) -\u003e HTTPTransportConfig | None:\n    \"\"\"Called automatically before each request with request details.\"\"\"\n    try:\n        token = auth_token_var.get()\n        headers = {\"Authorization\": f\"Bearer {token}\"}\n        \n        # Can make decisions based on request type/content\n        if isinstance(request.root, types.CallToolRequest):\n            headers[\"X-Tool-Name\"] = request.root.params.name\n        \n        return HTTPTransportConfig(headers=headers)\n    except LookupError:\n        return None\n\n\nasync def handle_requests_with_provider():\n    \"\"\"Transport config automatically added to all requests via provider.\"\"\"\n    async with streamable_http_client(\"https://mcp.example.com\") as (read, write, _):\n        async with ClientSession(\n            read,\n            write,\n            transport_config_provider=my_transport_provider,  # NEW\n        ) as session:\n            await session.initialize()\n            \n            # Context vars set here will be read by the provider\n            auth_token_var.set(\"user-123-token\")\n            result1 = await session.call_tool(\"tool1\", arguments={})\n            \n            auth_token_var.set(\"user-456-token\") \n            result2 = await session.call_tool(\"tool2\", arguments={})\n```\n\n### Transport Configuration Types (Option A)\n\n```python\n# Exported by mcp/client/streamable_http.py\n@dataclass\nclass HTTPTransportConfig:\n    \"\"\"Configuration for HTTP transport. Optional type hint for users.\"\"\"\n    headers: dict[str, str] | None = None\n    timeout: timedelta | None = None\n\n# Other transports can export their own configuration types\n```\n\n*Pros:* Simple, type hints available for IDE support, users can pass dict if they prefer  \n*Cons:* Provider code coupled to specific transport type (acceptable trade-off for simplicity)\n\n## Design Option B: Generic Context + Transport Mapper\n\nUser defines their own context type. Transport configuration requires a mapper function to convert to transport-specific format.\n\n### Example (Option B)\n\n```python\nfrom dataclasses import dataclass\nfrom mcp import ClientSession\nfrom mcp.client.streamable_http import streamable_http_client\n\n\n# User defines their own context type\n@dataclass\nclass MyAppContext:\n    user_token: str\n    trace_id: str\n\n\n# Transport mapper converts user context to transport format\ndef http_context_mapper(ctx: MyAppContext | None) -\u003e dict[str, Any] | None:\n    if ctx is None:\n        return None\n    return {\n        \"headers\": {\n            \"Authorization\": f\"Bearer {ctx.user_token}\",\n            \"X-Trace-ID\": ctx.trace_id,\n        }\n    }\n\n\n# Per-request usage\nasync def handle_single_request(session: ClientSession, user_token: str, trace_id: str):\n    result = await session.call_tool(\n        \"my_tool\",\n        arguments={\"foo\": \"bar\"},\n        request_context=MyAppContext(user_token=user_token, trace_id=trace_id)\n    )\n    return result\n```\n\nSession-level provider usage would work similarly to Option A.\n\n*Pros:* Strong contract, user owns context type, session-level code is transport-agnostic  \n*Cons:* Additional configuration, mapper required for transport to function\n\n---\n\n**Recommendation: Option A** for its simplicity. While it couples the provider to the transport type, this is acceptable since transport choice is typically fixed per deployment.\n\n## API Changes Summary\n\n### Session Changes\n\n```python\nclass TransportConfigProviderFnT(Protocol):\n    \"\"\"Provider callback that receives the request and returns transport configuration.\"\"\"\n    async def __call__(self, request: types.ClientRequest) -\u003e Any | None:\n        \"\"\"\n        Provide transport configuration for a request.\n        \n        Args:\n            request: The ClientRequest being sent (CallToolRequest, ListToolsRequest, etc.)\n                     Allows provider to inspect request type, method name, arguments, etc.\n        \n        Returns:\n            Transport configuration (typed or dict), or None for no configuration.\n        \"\"\"\n        ...\n\n\nclass ClientSession:\n    def __init__(\n        self,\n        # ... existing params ...\n        transport_config_provider: TransportConfigProviderFnT | None = None,  # NEW\n    ):\n        ...\n```\n\n### Method Changes (applies to call_tool, list_tools, etc.)\n\n```python\nasync def call_tool(\n    self,\n    name: str,\n    arguments: dict[str, Any] | None = None,\n    # ... existing params ...\n    transport_config: Any | None = None,  # NEW\n) -\u003e types.CallToolResult:\n    ...\n```\n\n**Note:** Transport configuration is separate from `RequestParams.Meta` (the `meta` parameter) which is protocol-level metadata serialized into the JSON-RPC message and sent to the MCP server. Transport configuration is consumed by the transport layer (e.g., HTTP headers) and never reaches the server.\n\n## Implementation Considerations\n\n### Backwards Compatibility\n- All new parameters should be optional with `None` defaults\n- Existing behavior unchanged when new features not used\n- Transport implementations can ignore `transport_config` if not supported\n\n### Deprecation\n- The existing `read_timeout_seconds` parameter on `call_tool()` should be deprecated in favor of passing timeout via `transport_config`\n\n### Functionality\n- Context provider callbacks should be async to support I/O (e.g., token refresh)\n\n## Related Issues\n- https://github.com/modelcontextprotocol/python-sdk/issues/1509\n\n\n### References\n\n_No response_","author":{"url":"https://github.com/somaraani","@type":"Person","name":"somaraani"},"datePublished":"2026-01-27T17:37:12.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":1},"url":"https://github.com/1966/python-sdk/issues/1966"}