modelcontextprotocol / modelcontextprotocol/python-sdk

Per-Request Transport Configuration for MCP Clients

Aperta
#1,966 2 commenti 1 reazione 0 assegnatari Vedi su GitHub

Nessuno ha ancora preso questa issue.

documentation enhancement v2
Lingua principale
Python
Stelle
24.3k
Fork
4k
Merge medio
1g 1h
PR unite (30g)
31

Descrizione

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

Guida per i contributori

Apri la guida per i contributori

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Direzione di ricerca

Inizia dai metodi di richiesta di ClientSession e mcp/client/streamable_http.py, in particolare dalla gestione esistente di read_timeout_seconds e dalla configurazione del trasporto. Traccia il percorso delle richieste fino al livello di trasporto e rivedi la issue 1509 correlata prima di scegliere tra i design di configurazione proposti. Il lavoro è completato quando la configurazione opzionale per richiesta e del provider della sessione funziona per tutte le richieste applicabili senza modificare il comportamento esistente e la migrazione del timeout è stata affrontata.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Valutazione

Stack tecnologico
python
Ambito
api, networking
Tipo di issue
Funzionalità
Difficoltà
5/5
Tempo stimato
Più di una settimana
Stato di attività
Tranquilla
Chiarezza
Abbastanza chiara
Idoneità per principianti
38/100

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.