modelcontextprotocol / modelcontextprotocol/python-sdk
Per-Request Transport Configuration for MCP Clients
还没有人认领这个 Issue。
- 主要语言
- 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:
- HTTP headers are configured on the
httpx.AsyncClientat connection establishment - Session-level context is fixed at
ClientSessioncreation - 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:
- Per-request - Pass configuration directly when calling a method
- 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
Nonedefaults - Existing behavior unchanged when new features not used
- Transport implementations can ignore
transport_configif not supported
Deprecation
- The existing
read_timeout_secondsparameter oncall_tool()should be deprecated in favor of passing timeout viatransport_config
Functionality
- Context provider callbacks should be async to support I/O (e.g., token refresh)
Related Issues
References
No response
贡献指南
从这里开始
- 先读完整个 Issue,再读项目的贡献指南。
- 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
- Fork 仓库,在一个分支上完成修改。
- 提交 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