modelcontextprotocol / modelcontextprotocol/python-sdk

Per-Request Transport Configuration for MCP Clients

未关闭
#1,966 2 条评论 1 个 reaction 已指派 0 人 在 GitHub 查看

还没有人认领这个 Issue。

documentation enhancement v2
主要语言
Python
星标
24.3k
派生
4k
平均合并
1 天 1 小时
30 天内合并 PR
31

描述

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

References

No response

贡献指南

打开贡献指南

从这里开始

  1. 先读完整个 Issue,再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库,在一个分支上完成修改。
  4. 提交 Pull Request,并在描述里引用这个 Issue 编号。

调研方向

从 ClientSession 的请求方法和 mcp/client/streamable_http.py 开始,重点查看现有的 read_timeout_seconds 处理方式和传输层设置。跟踪请求如何到达传输层,并在提出的配置设计中进行选择之前,查看相关的 issue 1509。完成的标准是:按请求和 session-provider 提供的可选配置能够适用于所有相关请求,同时不改变现有行为,并且已处理 timeout migration。

由索引模型根据 Issue 内容生成。

评估

技术栈
python
领域
api, networking
Issue 类型
功能
难度
5/5
预计耗时
一周以上
活跃度
冷清
描述清晰度
基本清楚
新手友好度
38/100

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。