Stop mcp.client from importing the server stack

Client only needs Server/MCPServer/InMemoryTransport/modern_on_request when it
is handed a live in-process server, and a Server instance can only exist once
the user has imported the server stack. Detect that case via sys.modules
instead of eager imports, keep the server types for static checking under
TYPE_CHECKING, and import the in-memory bridge inside _connect_inproc.

Author: maxisbey

src/mcp/client/client.py

@@ -4,11 +4,12 @@
 
 import hashlib
 import logging
+import sys
 import uuid
 from collections.abc import Awaitable, Callable, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, AsyncExitStack
 from dataclasses import KW_ONLY, dataclass, field
-from typing import Any, Literal, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast
 
 import anyio
 import anyio.lowlevel
@@ -40,10 +41,9 @@
     ServerCapabilities,
 )
 from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS, MODERN_PROTOCOL_VERSIONS
-from typing_extensions import deprecated
+from typing_extensions import TypeIs, deprecated
 
 from mcp.client._input_required import DEFAULT_INPUT_REQUIRED_MAX_ROUNDS, run_input_required_driver
-from mcp.client._memory import InMemoryTransport
 from mcp.client._probe import negotiate_auto
 from mcp.client._transport import Transport
 from mcp.client.caching import CacheConfig, CacheMode, ClientResponseCache, InMemoryResponseCacheStore
@@ -61,16 +61,20 @@
 from mcp.client.streamable_http import streamable_http_client
 from mcp.client.subscriptions import ServerEvent, Subscription
 from mcp.client.subscriptions import listen as _listen
-from mcp.server import Server
-from mcp.server.mcpserver import MCPServer
-from mcp.server.runner import modern_on_request
 from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
 from mcp.shared.dispatcher import Dispatcher, ProgressFnT
 from mcp.shared.exceptions import MCPDeprecationWarning, MCPError
 from mcp.shared.extension import validate_extension_identifier
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
 from mcp.shared.subscriptions import event_to_notification
 
+if TYPE_CHECKING:
+    # Typing-only: the server stack is never imported by the client at runtime.
+    # An in-process `Server`/`MCPServer` argument already implies the caller
+    # imported it (see `_is_lowlevel_server`); URL/`Transport` clients pay nothing.
+    from mcp.server import Server
+    from mcp.server.mcpserver import MCPServer
+
 logger = logging.getLogger(__name__)
 
 ConnectMode = Literal["legacy", "auto"] | str
@@ -99,10 +103,35 @@ async def connect(exit_stack: AsyncExitStack, _mode: ConnectMode, _raise_excepti
     return connect
 
 
+def _loaded_class(module: str, name: str) -> type | None:
+    """Return class `name` from `module` if that module has already been imported, else `None`.
+
+    A `Server`/`MCPServer` instance can only exist once its defining module has run, so
+    looking the class up in `sys.modules` (never importing it) gives an exact `isinstance`
+    test while keeping the whole server stack out of a transport-only client's imports.
+    """
+    return getattr(sys.modules.get(module), name, None)
+
+
+def _is_mcpserver(obj: object) -> TypeIs[MCPServer]:
+    cls = _loaded_class("mcp.server.mcpserver.server", "MCPServer")
+    return cls is not None and isinstance(obj, cls)
+
+
+def _is_lowlevel_server(obj: object) -> TypeIs[Server[Any]]:
+    cls = _loaded_class("mcp.server.lowlevel.server", "Server")
+    return cls is not None and isinstance(obj, cls)
+
+
 def _connect_inproc(server: Server[Any]) -> _Connector:
     """Connector for an in-process ``Server``: legacy mode drives the stream loop via
     ``InMemoryTransport``; any other mode drives the modern per-request path through a
     ``DirectDispatcher`` peer pair (no streams, no JSON-RPC framing, no initialize handshake)."""
+    # Function-level imports (documented exception): this connector is only built for a live
+    # in-process `Server`, whose stack the caller has therefore already imported; binding these
+    # here rather than at module top keeps `import mcp.client` free of the server stack.
+    from mcp.client._memory import InMemoryTransport
+    from mcp.server.runner import modern_on_request
 
     async def connect(exit_stack: AsyncExitStack, mode: ConnectMode, raise_exceptions: bool) -> Dispatcher[Any]:
         if mode == "legacy":
@@ -388,9 +417,9 @@ def __post_init__(self) -> None:
         self._folded_extensions = _fold_extensions(self.extensions)
 
         srv = self.server
-        if isinstance(srv, MCPServer):
+        if _is_mcpserver(srv):
             srv = srv._lowlevel_server  # pyright: ignore[reportPrivateUsage]
-        if isinstance(srv, Server):
+        if _is_lowlevel_server(srv):
             self._connect = _connect_inproc(srv)
         elif isinstance(srv, str):
             self._connect = _connect_transport(streamable_http_client(srv))