Ensure client is only initialized once

src/fastmcp/client/client.py

@@ -145,6 +145,7 @@ class Client(Generic[ClientTransportT]):
         progress_handler: ProgressHandler | None = None,
         timeout: datetime.timedelta | float | int | None = None,
         init_timeout: datetime.timedelta | float | int | None = None,
+        client_info: mcp.types.Implementation | None = None,
         auth: httpx.Auth | Literal["oauth"] | str | None = None,
     ):
         self.transport = cast(ClientTransportT, infer_transport(transport))
@@ -180,6 +181,7 @@ class Client(Generic[ClientTransportT]):
             "logging_callback": create_log_callback(log_handler),
             "message_handler": message_handler,
             "read_timeout_seconds": timeout,
+            "client_info": client_info,
         }
 
         if roots is not None:

src/fastmcp/client/transports.py

@@ -8,27 +8,15 @@ import sys
 import warnings
 from collections.abc import AsyncIterator, Callable
 from pathlib import Path
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Literal,
-    TypedDict,
-    TypeVar,
-    cast,
-    overload,
-)
+from typing import TYPE_CHECKING, Any, Literal, TypedDict, TypeVar, cast, overload
 
 import anyio
 import httpx
+import mcp.types
 from mcp import ClientSession, StdioServerParameters
-from mcp.client.session import (
-    ListRootsFnT,
-    LoggingFnT,
-    MessageHandlerFnT,
-    SamplingFnT,
-)
+from mcp.client.session import ListRootsFnT, LoggingFnT, MessageHandlerFnT, SamplingFnT
 from mcp.server.fastmcp import FastMCP as FastMCP1Server
-from mcp.shared.memory import create_connected_server_and_client_session
+from mcp.shared.memory import create_client_server_memory_streams
 from pydantic import AnyUrl
 from typing_extensions import Unpack
 
@@ -65,11 +53,12 @@ __all__ = [
 class SessionKwargs(TypedDict, total=False):
     """Keyword arguments for the MCP ClientSession constructor."""
 
+    read_timeout_seconds: datetime.timedelta | None
     sampling_callback: SamplingFnT | None
     list_roots_callback: ListRootsFnT | None
     logging_callback: LoggingFnT | None
     message_handler: MessageHandlerFnT | None
-    read_timeout_seconds: datetime.timedelta | None
+    client_info: mcp.types.Implementation | None
 
 
 class ClientTransport(abc.ABC):
@@ -662,24 +651,47 @@ class FastMCPTransport(ClientTransport):
     tests or scenarios where client and server run in the same runtime.
     """
 
-    def __init__(self, mcp: FastMCP | FastMCP1Server):
+    def __init__(self, mcp: FastMCP | FastMCP1Server, raise_exceptions: bool = False):
         """Initialize a FastMCPTransport from a FastMCP server instance."""
 
         # Accept both FastMCP 2.x and FastMCP 1.0 servers. Both expose a
         # ``_mcp_server`` attribute pointing to the underlying MCP server
         # implementation, so we can treat them identically.
         self.server = mcp
+        self.raise_exceptions = raise_exceptions
 
     @contextlib.asynccontextmanager
     async def connect_session(
         self, **session_kwargs: Unpack[SessionKwargs]
     ) -> AsyncIterator[ClientSession]:
-        # create_connected_server_and_client_session manages the session lifecycle itself
-        async with create_connected_server_and_client_session(
-            server=self.server._mcp_server,
-            **session_kwargs,
-        ) as session:
-            yield session
+        async with create_client_server_memory_streams() as (
+            client_streams,
+            server_streams,
+        ):
+            client_read, client_write = client_streams
+            server_read, server_write = server_streams
+
+            # Create a cancel scope for the server task
+            async with anyio.create_task_group() as tg:
+                tg.start_soon(
+                    lambda: self.server._mcp_server.run(
+                        server_read,
+                        server_write,
+                        self.server._mcp_server.create_initialization_options(),
+                        raise_exceptions=self.raise_exceptions,
+                    )
+                )
+
+                try:
+                    async with ClientSession(
+                        read_stream=client_read,
+                        write_stream=client_write,
+                        **session_kwargs,
+                        client_info=session_kwargs.get("client_info"),
+                    ) as client_session:
+                        yield client_session
+                finally:
+                    tg.cancel_scope.cancel()
 
     def __repr__(self) -> str:
         return f"<FastMCPTransport(server='{self.server.name}')>"