Stop proxies from validating backend results or mutating shared transports

docs/development/v4-notes/change-register.mdx

@@ -205,6 +205,31 @@ client = Client("my_mcp_server.py", timeout=30.0)  # also works
 
 *Verify:* `fastmcp_slim/fastmcp/client/transports/base.py` (`SessionKwargs.read_timeout_seconds: float | None`), `client/client.py`.
 
+### Connection settings passed to `connect_session` — Breaking (custom transports)
+
+`ClientTransport.connect_session` takes a new keyword-only `transport_options: TransportOptions | None`, describing how the connecting client wants its session built: which `ClientSession` class to instantiate, and whether to forward the caller's authorization header upstream. Proxies use it to relay backend results without enforcing their output schema (see [Proxy Servers](/servers/providers/proxy#tool-results-are-relayed-not-inspected)).
+
+These settings previously lived on the transport instance, so a transport shared between clients leaked one client's configuration into another — including credential forwarding, which `create_proxy(some_client)` would silently enable on the caller's own client. They now travel with the client that wants them, and `forward_incoming_headers` is no longer a settable transport attribute.
+
+A client only passes the argument when it wants non-default settings, so an ordinary `Client` is unaffected and transports that don't accept it keep working. A custom `ClientTransport` used as a *proxy backend* must accept and honor it:
+
+```python
+import contextlib
+
+from fastmcp.client.transports.base import ClientTransport, TransportOptions
+
+class MyTransport(ClientTransport):
+    @contextlib.asynccontextmanager
+    async def connect_session(self, *, transport_options=None, **session_kwargs):
+        options = transport_options or TransportOptions()
+        async with options.session_class(read, write, **session_kwargs) as session:
+            yield session
+```
+
+A transport that wraps others must pass it along; `MCPConfigTransport` forwards it to both its single-server delegate and its composite server.
+
+*Verify:* `fastmcp_slim/fastmcp/client/transports/base.py` (`TransportOptions`), the four built-in transports, `transports/config.py`, and `tests/server/providers/proxy/test_proxy_server.py`.
+
 ### `get_session_id` via header sniff — Bridged
 
 The SDK dropped `get_session_id` from the streamable-HTTP transport with no replacement (the SDK source has an author TODO acknowledging it breaks the Transport protocol). FastMCP reconstructs it by registering an httpx2 response event hook on the client it owns, capturing the `mcp-session-id` response header (httpx2 preserves httpx's `event_hooks` API). The removal trigger is the upstream TODO.

docs/servers/providers/proxy.mdx

@@ -170,6 +170,26 @@ backend = ProxyClient(
 )
 ```
 
+### Tool Results Are Relayed, Not Inspected
+
+A proxy passes a backend's tool results through untouched, including results that don't match the output schema the backend advertised. Deciding whether a server honored its own contract belongs to the client consuming the result, and that client validates for itself.
+
+This matters when a backend's declared schema is subtly wrong — an enum missing a variant it actually returns, say. A proxy that enforced the schema would replace the backend's working response with an error of its own, and the client would never see what the backend actually said.
+
+```python
+from fastmcp import Client
+from fastmcp.server import create_proxy
+
+proxy = create_proxy("backend_server.py")
+
+async with Client(proxy) as client:
+    # The backend's response arrives as the backend sent it. If it violates
+    # the backend's own output schema, this client raises — its decision.
+    result = await client.call_tool("get_status")
+```
+
+Skipping the check also avoids a `tools/list` round trip to the backend on every proxied call, since validation would need the backend's schemas and a proxy builds a fresh connection per request.
+
 ## Configuration-Based Proxies
 
 <VersionBadge version="2.4.0" />

fastmcp_slim/fastmcp/client/client.py

@@ -98,6 +98,7 @@
     StreamableHttpTransport,
     infer_transport,
 )
+from .transports.base import TransportOptions
 
 __all__ = [
     "Client",
@@ -490,6 +491,7 @@ def __init__(
 
         # Session context management - see class docstring for detailed explanation
         self._session_state = ClientSessionState()
+        self._transport_options: TransportOptions | None = None
 
         # Track task IDs submitted by this client (for list_tasks support)
         self._submitted_task_ids: set[str] = set()
@@ -656,6 +658,7 @@ def new(self) -> Client[ClientTransportT]:
         # Always reset session state so cloned clients start disconnected and do not
         # share lifecycle state with the original instance.
         new_client._session_state = ClientSessionState()
+        new_client._transport_options = self._transport_options
 
         # Reset mutable task tracking state so new client is independent
         new_client._task_registry = {}
@@ -695,10 +698,17 @@ def new(self) -> Client[ClientTransportT]:
 
     @asynccontextmanager
     async def _context_manager(self):
+        # Only passed when this client actually wants non-default settings, so an
+        # ordinary client never sends an argument a transport might not accept.
+        if self._transport_options is not None:
+            connection = self.transport.connect_session(
+                transport_options=self._transport_options, **self._session_kwargs
+            )
+        else:
+            connection = self.transport.connect_session(**self._session_kwargs)
+
         with catch(get_catch_handlers()):
-            async with self.transport.connect_session(
-                **self._session_kwargs
-            ) as session:
+            async with connection as session:
                 self._session_state.session = session
                 # Initialize the session if auto_initialize is enabled
                 try:

fastmcp_slim/fastmcp/client/transports/base.py

@@ -1,6 +1,7 @@
 import abc
 import contextlib
 from collections.abc import AsyncIterator, Sequence
+from dataclasses import dataclass
 from typing import Any, Literal, TypeVar
 
 import httpx2
@@ -20,7 +21,7 @@
 ClientTransportT = TypeVar("ClientTransportT", bound="ClientTransport")
 
 
-class SessionKwargs(TypedDict, total=False):
+class ClientSessionKwargs(TypedDict, total=False):
     """Keyword arguments for the MCP ClientSession constructor."""
 
     read_timeout_seconds: float | None
@@ -34,6 +35,32 @@ class SessionKwargs(TypedDict, total=False):
     notification_bindings: Sequence[NotificationBinding[Any]] | None
 
 
+@dataclass(frozen=True)
+class TransportOptions:
+    """How one client wants its connection built.
+
+    These belong to the client rather than to the transport, so a transport
+    shared between clients doesn't leak one client's settings to another.
+
+    Attributes:
+        session_class: The ClientSession class to instantiate. Proxies supply a
+            session that skips output-schema validation, since they relay
+            results rather than consume them.
+        forward_incoming_headers: Whether to forward the inbound request's
+            authorization header upstream. Only appropriate for proxies, where
+            the caller's credentials are meant to be propagated. Honored by the
+            HTTP and SSE transports; ignored by the others.
+    """
+
+    session_class: type[ClientSession] = ClientSession
+    forward_incoming_headers: bool = False
+
+
+# SessionKwargs stays exactly the ClientSession constructor's parameters, so a
+# transport can splat it into ClientSession without filtering.
+SessionKwargs = ClientSessionKwargs
+
+
 class ClientTransport(abc.ABC):
     """
     Abstract base class for different MCP client transport mechanisms.
@@ -46,7 +73,10 @@ class ClientTransport(abc.ABC):
     @abc.abstractmethod
     @contextlib.asynccontextmanager
     async def connect_session(
-        self, **session_kwargs: Unpack[SessionKwargs]
+        self,
+        *,
+        transport_options: TransportOptions | None = None,
+        **session_kwargs: Unpack[SessionKwargs],
     ) -> AsyncIterator[ClientSession]:
         """
         Establishes a connection and yields an active ClientSession.
@@ -58,6 +88,9 @@ async def connect_session(
         within this context.
 
         Args:
+            transport_options: How the connecting client wants this connection
+                               built. Defaults apply when omitted. A transport
+                               that wraps others must pass this along.
             **session_kwargs: Keyword arguments to pass to the ClientSession
                               constructor (e.g., callbacks, timeouts).
 

fastmcp_slim/fastmcp/client/transports/config.py

@@ -6,7 +6,11 @@
 from typing_extensions import Unpack
 
 from fastmcp import _install_hints
-from fastmcp.client.transports.base import ClientTransport, SessionKwargs
+from fastmcp.client.transports.base import (
+    ClientTransport,
+    SessionKwargs,
+    TransportOptions,
+)
 from fastmcp.client.transports.memory import FastMCPTransport
 from fastmcp.mcp_config import (
     MCPConfig,
@@ -89,11 +93,16 @@ def __init__(self, config: MCPConfig | dict, name_as_prefix: bool = True):
 
     @contextlib.asynccontextmanager
     async def connect_session(
-        self, **session_kwargs: Unpack[SessionKwargs]
+        self,
+        *,
+        transport_options: TransportOptions | None = None,
+        **session_kwargs: Unpack[SessionKwargs],
     ) -> AsyncIterator[ClientSession]:
         # Single server - delegate directly to pre-created transport
         if len(self.config.mcpServers) == 1:
-            async with self.transport.connect_session(**session_kwargs) as session:
+            async with self.transport.connect_session(
+                transport_options=transport_options, **session_kwargs
+            ) as session:
                 yield session
             return
 
@@ -138,7 +147,7 @@ async def connect_session(
                 raise ConnectionError("All MCP servers failed to connect")
 
             async with FastMCPTransport(mcp=composite).connect_session(
-                **session_kwargs
+                transport_options=transport_options, **session_kwargs
             ) as session:
                 yield session
 

fastmcp_slim/fastmcp/client/transports/http.py

@@ -17,7 +17,11 @@
 from fastmcp.client.auth.bearer import BearerAuth
 from fastmcp.client.auth.oauth import OAuth
 from fastmcp.client.dependencies import get_http_headers
-from fastmcp.client.transports.base import ClientTransport, SessionKwargs
+from fastmcp.client.transports.base import (
+    ClientTransport,
+    SessionKwargs,
+    TransportOptions,
+)
 
 
 class StreamableHttpTransport(ClientTransport):
@@ -74,8 +78,6 @@ def __init__(
 
         self._set_auth(auth)
 
-        self.forward_incoming_headers: bool = False
-
         # SDK v2's streamable_http_client no longer exposes a get_session_id
         # callback. We recover the session id ourselves by capturing the
         # `mcp-session-id` response header via an httpx event hook on the
@@ -143,13 +145,18 @@ def factory(
 
     @contextlib.asynccontextmanager
     async def connect_session(
-        self, **session_kwargs: Unpack[SessionKwargs]
+        self,
+        *,
+        transport_options: TransportOptions | None = None,
+        **session_kwargs: Unpack[SessionKwargs],
     ) -> AsyncIterator[ClientSession]:
+        options = transport_options or TransportOptions()
+
         # When used in a proxy, forward the inbound request's authorization
         # header to the upstream server. This is off by default so that a
         # plain Client used inside a server tool handler doesn't accidentally
         # leak the caller's credentials to an unrelated remote server.
-        if self.forward_incoming_headers:
+        if options.forward_incoming_headers:
             headers = get_http_headers(include={"authorization"}) | self.headers
         else:
             headers = dict(self.headers)
@@ -202,7 +209,9 @@ async def connect_session(
                 read_stream,
                 write_stream,
             ),
-            ClientSession(read_stream, write_stream, **session_kwargs) as session,
+            options.session_class(
+                read_stream, write_stream, **session_kwargs
+            ) as session,
         ):
             yield session
 

fastmcp_slim/fastmcp/client/transports/memory.py

@@ -11,7 +11,11 @@
 from typing_extensions import Unpack
 
 from fastmcp import _install_hints
-from fastmcp.client.transports.base import ClientTransport, SessionKwargs
+from fastmcp.client.transports.base import (
+    ClientTransport,
+    SessionKwargs,
+    TransportOptions,
+)
 
 if TYPE_CHECKING:
     from fastmcp.server.server import FastMCP
@@ -52,8 +56,12 @@ def __init__(self, mcp: "FastMCP[Any] | SDKServer", raise_exceptions: bool = Fal
 
     @contextlib.asynccontextmanager
     async def connect_session(
-        self, **session_kwargs: Unpack[SessionKwargs]
+        self,
+        *,
+        transport_options: TransportOptions | None = None,
+        **session_kwargs: Unpack[SessionKwargs],
     ) -> AsyncIterator[ClientSession]:
+        options = transport_options or TransportOptions()
         async with create_client_server_memory_streams() as (
             client_streams,
             server_streams,
@@ -88,7 +96,7 @@ async def connect_session(
                     )
 
                     try:
-                        async with ClientSession(
+                        async with options.session_class(
                             read_stream=client_read,
                             write_stream=client_write,
                             **session_kwargs,

fastmcp_slim/fastmcp/client/transports/sse.py

@@ -18,7 +18,11 @@
 from fastmcp.client.auth.bearer import BearerAuth
 from fastmcp.client.auth.oauth import OAuth
 from fastmcp.client.dependencies import get_http_headers
-from fastmcp.client.transports.base import ClientTransport, SessionKwargs
+from fastmcp.client.transports.base import (
+    ClientTransport,
+    SessionKwargs,
+    TransportOptions,
+)
 from fastmcp.utilities.timeout import normalize_timeout_to_timedelta
 
 
@@ -61,8 +65,6 @@ def __init__(
 
         self._set_auth(auth)
 
-        self.forward_incoming_headers: bool = False
-
         self.sse_read_timeout = normalize_timeout_to_timedelta(sse_read_timeout)
 
     def _set_auth(self, auth: httpx2.Auth | Literal["oauth"] | str | None):
@@ -115,15 +117,19 @@ def factory(
 
     @contextlib.asynccontextmanager
     async def connect_session(
-        self, **session_kwargs: Unpack[SessionKwargs]
+        self,
+        *,
+        transport_options: TransportOptions | None = None,
+        **session_kwargs: Unpack[SessionKwargs],
     ) -> AsyncIterator[ClientSession]:
+        options = transport_options or TransportOptions()
         client_kwargs: dict[str, Any] = {}
 
         # When used in a proxy, forward the inbound request's authorization
         # header to the upstream server. This is off by default so that a
         # plain Client used inside a server tool handler doesn't accidentally
         # leak the caller's credentials to an unrelated remote server.
-        if self.forward_incoming_headers:
+        if options.forward_incoming_headers:
             client_kwargs["headers"] = (
                 get_http_headers(include={"authorization"}) | self.headers
             )
@@ -149,7 +155,7 @@ async def connect_session(
 
         async with sse_client(self.url, auth=self.auth, **client_kwargs) as transport:
             read_stream, write_stream = transport
-            async with ClientSession(
+            async with options.session_class(
                 read_stream, write_stream, **session_kwargs
             ) as session:
                 yield session