Per-Request Transport Configuration for MCP Clients

[somaraani]

Summary

Add 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.

Problem Statement

Currently, MCP clients only supports setting context at initialization time:

1. HTTP headers are configured on the httpx.AsyncClient at connection establishment
2. Session-level context is fixed at ClientSession creation
3. Transport-level configuration is determined when the transport context manager is entered

This architecture doesn't support common production requirements:

• Per-Request Authentication: Auth tokens that need to be propagated per-request based on the calling user's context
• Request Tracing: Trace IDs/span IDs that change per request
• Dynamic Headers: Headers that vary based on request context (tenant IDs, correlation IDs, etc.)

Current Workaround Attempts

Users attempting to propagate context via contextvars face challenges:

• Context vars are copied at session start, not per-request
• The background thread pattern means context vars set after initialization are not visible
• Even with context propagation to the background thread, there's no mechanism to pass that context to the HTTP transport layer

Current API Limitations

The 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.

In addition, this functionality should extend beyond just tool calls, as headers or timeout configurations may be required for all requests to the server.

Proposed Solution

There are two levels where context can be provided:

1. Per-request - Pass configuration directly when calling a method
2. Session-level provider - A callback that provides configuration automatically for every request

For both levels, there are two design approaches for how configuration flows to the transport.

Design Option A: Transport-Specific Configuration (Recommended)

Each transport exports its own configuration type. Users provide settings in the transport's expected format.

Complete Example (Option A)

from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client, HTTPTransportConfig

# === Per-Request Usage ===
async def handle_single_request(session: ClientSession, user_token: str):
    """Pass transport config directly on each call."""
    result = await session.call_tool(
        "my_tool",
        arguments={"foo": "bar"},
        # Pass transport config directly - can use typed class or dict
        transport_config=HTTPTransportConfig(
            headers={"Authorization": f"Bearer {user_token}"}
        )
        # Or simply: transport_config={"headers": {"Authorization": f"Bearer {user_token}"}}
    )
    return result


# === Session-Level Provider Usage ===
async def my_transport_provider(request: types.ClientRequest) -> HTTPTransportConfig | None:
    """Called automatically before each request with request details."""
    try:
        token = auth_token_var.get()
        headers = {"Authorization": f"Bearer {token}"}
        
        # Can make decisions based on request type/content
        if isinstance(request.root, types.CallToolRequest):
            headers["X-Tool-Name"] = request.root.params.name
        
        return HTTPTransportConfig(headers=headers)
    except LookupError:
        return None


async def handle_requests_with_provider():
    """Transport config automatically added to all requests via provider."""
    async with streamable_http_client("https://mcp.example.com") as (read, write, _):
        async with ClientSession(
            read,
            write,
            transport_config_provider=my_transport_provider,  # NEW
        ) as session:
            await session.initialize()
            
            # Context vars set here will be read by the provider
            auth_token_var.set("user-123-token")
            result1 = await session.call_tool("tool1", arguments={})
            
            auth_token_var.set("user-456-token") 
            result2 = await session.call_tool("tool2", arguments={})

Transport Configuration Types (Option A)

# Exported by mcp/client/streamable_http.py
@dataclass
class HTTPTransportConfig:
    """Configuration for HTTP transport. Optional type hint for users."""
    headers: dict[str, str] | None = None
    timeout: timedelta | None = None

# Other transports can export their own configuration types

Pros: Simple, type hints available for IDE support, users can pass dict if they prefer
Cons: Provider code coupled to specific transport type (acceptable trade-off for simplicity)

Design Option B: Generic Context + Transport Mapper

User defines their own context type. Transport configuration requires a mapper function to convert to transport-specific format.

Example (Option B)

from dataclasses import dataclass
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client


# User defines their own context type
@dataclass
class MyAppContext:
    user_token: str
    trace_id: str


# Transport mapper converts user context to transport format
def http_context_mapper(ctx: MyAppContext | None) -> dict[str, Any] | None:
    if ctx is None:
        return None
    return {
        "headers": {
            "Authorization": f"Bearer {ctx.user_token}",
            "X-Trace-ID": ctx.trace_id,
        }
    }


# Per-request usage
async def handle_single_request(session: ClientSession, user_token: str, trace_id: str):
    result = await session.call_tool(
        "my_tool",
        arguments={"foo": "bar"},
        request_context=MyAppContext(user_token=user_token, trace_id=trace_id)
    )
    return result

Session-level provider usage would work similarly to Option A.

Pros: Strong contract, user owns context type, session-level code is transport-agnostic
Cons: Additional configuration, mapper required for transport to function

---

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.

API Changes Summary

Session Changes

class TransportConfigProviderFnT(Protocol):
    """Provider callback that receives the request and returns transport configuration."""
    async def __call__(self, request: types.ClientRequest) -> Any | None:
        """
        Provide transport configuration for a request.
        
        Args:
            request: The ClientRequest being sent (CallToolRequest, ListToolsRequest, etc.)
                     Allows provider to inspect request type, method name, arguments, etc.
        
        Returns:
            Transport configuration (typed or dict), or None for no configuration.
        """
        ...


class ClientSession:
    def __init__(
        self,
        # ... existing params ...
        transport_config_provider: TransportConfigProviderFnT | None = None,  # NEW
    ):
        ...

Method Changes (applies to call_tool, list_tools, etc.)

async def call_tool(
    self,
    name: str,
    arguments: dict[str, Any] | None = None,
    # ... existing params ...
    transport_config: Any | None = None,  # NEW
) -> types.CallToolResult:
    ...

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.

Implementation Considerations

Backwards Compatibility

• All new parameters should be optional with None defaults
• Existing behavior unchanged when new features not used
• Transport implementations can ignore transport_config if not supported

Deprecation

• The existing read_timeout_seconds parameter on call_tool() should be deprecated in favor of passing timeout via transport_config

Functionality

• Context provider callbacks should be async to support I/O (e.g., token refresh)

Related Issues

• Support Per-Request HTTP Headers in call_tool() #1509

References

No response