Fix concurrent proxy client operations with session isolation (#1083)

* Refactor Client session state with ClientSessionState dataclass

Fixes #1068 by introducing ClientSessionState to encapsulate session management
attributes, simplifying Client.new() and preventing concurrent proxy client
context mixing through client factory pattern.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

* Update proxy documentation for new client factory pattern

Documents the new session management behavior, client_factory parameter,
and concurrent operation safety introduced in v2.10.3 to fix #1068.

* Remove deprecation of client parameter and update documentation

- Undeprecated FastMCPProxy client parameter
- Made client_factory the advanced option for custom control
- Added detailed explanation of how client factories work internally
- Removed outdated note about proxy feature limitations
- Removed fabricated Advanced Usage section

* Re-do proxy documentation

* Fix deprecated client parameter to provide session isolation

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>

docs/servers/proxy.mdx

@@ -10,13 +10,10 @@ import { VersionBadge } from '/snippets/version-badge.mdx'
 
 FastMCP provides a powerful proxying capability that allows one FastMCP server instance to act as a frontend for another MCP server (which could be remote, running on a different transport, or even another FastMCP instance). This is achieved using the `FastMCP.as_proxy()` class method.
 
-`as_proxy()` accepts either an existing `Client` or any argument that can be passed to a `Client` as its `transport` parameter—such as another `FastMCP` instance, a URL to a remote server, or an MCP configuration dictionary.
-
 ## What is Proxying?
 
 Proxying means setting up a FastMCP server that doesn't implement its own tools or resources directly. Instead, when it receives a request (like `tools/call` or `resources/read`), it forwards that request to a *backend* MCP server, receives the response, and then relays that response back to the original client.
 
-
 ```mermaid
 sequenceDiagram
     participant ClientApp as Your Client (e.g., Claude Desktop)
@@ -30,81 +27,162 @@ sequenceDiagram
     Note over ClientApp, FastMCPProxy: Proxy relays the response
     FastMCPProxy-->>ClientApp: MCP Response (e.g. stdio)
 ```
-### Use Cases
 
--   **Transport Bridging**: Expose a server running on one transport (e.g., a remote SSE server) via a different transport (e.g., local Stdio for Claude Desktop).
--   **Adding Functionality**: Insert a layer in front of an existing server to add caching, logging, authentication, or modify requests/responses (though direct modification requires subclassing `FastMCPProxy`).
--   **Security Boundary**: Use the proxy as a controlled gateway to an internal server.
--   **Simplifying Client Configuration**: Provide a single, stable endpoint (the proxy) even if the backend server's location or transport changes.
+### Key Benefits
+
+<VersionBadge version="2.10.3" />
+
+- **Session Isolation**: Each request gets its own isolated session, ensuring safe concurrent operations
+- **Transport Bridging**: Expose servers running on one transport via a different transport
+- **Advanced MCP Features**: Automatic forwarding of sampling, elicitation, logging, and progress
+- **Security**: Acts as a controlled gateway to backend servers
+- **Simplicity**: Single endpoint even if backend location or transport changes
+
+## Quick Start
 
-## Creating a Proxy
+<VersionBadge version="2.10.3" />
 
-The easiest way to create a proxy is using the `FastMCP.as_proxy()` class method. This creates a standard FastMCP server that forwards requests to another MCP server.
+The recommended way to create a proxy is using `ProxyClient`, which provides full MCP feature support with automatic session isolation:
 
 ```python
 from fastmcp import FastMCP
+from fastmcp.server.proxy import ProxyClient
 
-# Provide the backend in any form accepted by Client
-proxy_server = FastMCP.as_proxy(
-    "backend_server.py",  # Could also be a FastMCP instance, config dict, or a remote URL
-    name="MyProxyServer"  # Optional settings for the proxy
+# Create a proxy with full MCP feature support
+proxy = FastMCP.as_proxy(
+    ProxyClient("backend_server.py"),
+    name="MyProxy"
 )
 
-# Or create the Client yourself for custom configuration
-backend_client = Client("backend_server.py")
-proxy_from_client = FastMCP.as_proxy(backend_client)
+# Run the proxy (e.g., via stdio for Claude Desktop)
+if __name__ == "__main__":
+    proxy.run()
 ```
 
-**How `as_proxy` Works:**
+This single setup gives you:
+- Safe concurrent request handling
+- Automatic forwarding of advanced MCP features (sampling, elicitation, etc.)
+- Session isolation to prevent context mixing
+- Full compatibility with all MCP clients
 
-1.  It connects to the backend server using the provided client.
-2.  It discovers all the tools, resources, resource templates, and prompts available on the backend server.
-3.  It creates corresponding "proxy" components that forward requests to the backend.
-4.  It returns a standard `FastMCP` server instance that can be used like any other.
+## Session Isolation & Concurrency
 
-<Note>
-Currently, proxying focuses primarily on exposing the major MCP objects (tools, resources, templates, and prompts). Some advanced MCP features like notifications and sampling are not fully supported in proxies in the current version. Support for these additional features may be added in future releases.
-</Note>
+<VersionBadge version="2.10.3" />
+
+FastMCP proxies provide session isolation to ensure safe concurrent operations. The session strategy depends on how the proxy is configured:
 
-### Bridging Transports
+### Fresh Sessions
 
-A common use case is to bridge transports. For example, making a remote SSE server available locally via Stdio:
+When you pass a disconnected client (which is the normal case), each request gets its own isolated backend session:
 
 ```python
-from fastmcp import FastMCP
+from fastmcp.server.proxy import ProxyClient
 
-# Target a remote SSE server directly by URL
-proxy = FastMCP.as_proxy("http://example.com/mcp/sse", name="SSE to Stdio Proxy")
+# Each request creates a fresh backend session (recommended)
+proxy = FastMCP.as_proxy(ProxyClient("backend_server.py"))
 
-# The proxy can now be used with any transport
-# No special handling needed - it works like any FastMCP server
+# Multiple clients can use this proxy simultaneously without interference:
+# - Client A calls a tool -> gets isolated backend session
+# - Client B calls a tool -> gets different isolated backend session  
+# - No context mixing between requests
 ```
 
-### In-Memory Proxies
+### Session Reuse with Connected Clients
+
+When you pass an already-connected client, the proxy will reuse that session for all requests:
+
+```python
+from fastmcp import Client
+
+# Create and connect a client
+async with Client("backend_server.py") as connected_client:
+    # This proxy will reuse the connected session for all requests
+    proxy = FastMCP.as_proxy(connected_client)
+    
+    # ⚠️ Warning: All requests share the same backend session
+    # This may cause context mixing in concurrent scenarios
+```
+
+**Important**: Using shared sessions with concurrent requests from multiple clients may lead to context mixing and race conditions. This approach should only be used in single-threaded scenarios or when you have explicit synchronization.
+
+## Transport Bridging
 
-You can also proxy an in-memory `FastMCP` instance, which is useful for adjusting the configuration or behavior of a server you don't completely control.
+A common use case is bridging transports - exposing a server running on one transport via a different transport. For example, making a remote SSE server available locally via stdio:
 
 ```python
 from fastmcp import FastMCP
+from fastmcp.server.proxy import ProxyClient
 
-# Original server
-original_server = FastMCP(name="Original")
+# Bridge remote SSE server to local stdio
+remote_proxy = FastMCP.as_proxy(
+    ProxyClient("http://example.com/mcp/sse"),
+    name="Remote-to-Local Bridge"
+)
 
-@original_server.tool
-def tool_a() -> str: 
-    return "A"
+# Run locally via stdio for Claude Desktop
+if __name__ == "__main__":
+    remote_proxy.run()  # Defaults to stdio transport
+```
 
-# Create a proxy of the original server directly
-proxy = FastMCP.as_proxy(
-    original_server,
-    name="Proxy Server"
+Or expose a local server via HTTP for remote access:
+
+```python
+# Bridge local server to HTTP
+local_proxy = FastMCP.as_proxy(
+    ProxyClient("local_server.py"),
+    name="Local-to-HTTP Bridge"
 )
 
-# proxy is now a regular FastMCP server that forwards
-# requests to original_server
+# Run via HTTP for remote clients
+if __name__ == "__main__":
+    local_proxy.run(transport="http", host="0.0.0.0", port=8080)
+```
+
+
+## Advanced MCP Features
+
+<VersionBadge version="2.10.3" />
+
+`ProxyClient` automatically forwards advanced MCP protocol features between the backend server and clients connected to the proxy, ensuring full MCP compatibility.
+
+### Supported Features
+
+- **Roots**: Forwards filesystem root access requests to the client
+- **Sampling**: Forwards LLM completion requests from backend to client  
+- **Elicitation**: Forwards user input requests to the client
+- **Logging**: Forwards log messages from backend through to client
+- **Progress**: Forwards progress notifications during long operations
+
+```python
+from fastmcp.server.proxy import ProxyClient
+
+# ProxyClient automatically handles all these features
+backend = ProxyClient("advanced_backend.py")
+proxy = FastMCP.as_proxy(backend)
+
+# When the backend server:
+# - Requests LLM sampling -> forwarded to your client
+# - Logs messages -> appear in your client
+# - Reports progress -> shown in your client
+# - Needs user input -> prompts your client
 ```
 
-### Configuration-Based Proxies
+### Customizing Feature Support
+
+You can selectively disable forwarding by passing `None` for specific handlers:
+
+```python
+# Disable sampling but keep other features
+backend = ProxyClient(
+    "backend_server.py",
+    sampling_handler=None,  # Disable LLM sampling forwarding
+    log_handler=None        # Disable log forwarding
+)
+```
+
+When you use a transport string directly with `FastMCP.as_proxy()`, it automatically creates a `ProxyClient` internally to ensure full feature support.
+
+## Configuration-Based Proxies
 
 <VersionBadge version="2.4.0" />
 
@@ -123,7 +201,7 @@ config = {
     }
 }
 
-# Create a proxy to the configured server
+# Create a proxy to the configured server (auto-creates ProxyClient)
 proxy = FastMCP.as_proxy(config, name="Config-Based Proxy")
 
 # Run the proxy with stdio transport for local access
@@ -135,11 +213,11 @@ if __name__ == "__main__":
 The MCPConfig format follows an emerging standard for MCP server configuration and may evolve as the specification matures. While FastMCP aims to maintain compatibility with future versions, be aware that field names or structure might change.
 </Note>
 
-You can also use MCPConfig to create a proxy to multiple servers. When multiple servers are specified, they are automatically mounted with their config names as prefixes, providing a unified interface to all servers:
+### Multi-Server Configurations
 
-```python
-from fastmcp import FastMCP
+You can create a proxy to multiple servers by specifying multiple entries in the config. They are automatically mounted with their config names as prefixes:
 
+```python
 # Multi-server configuration
 config = {
     "mcpServers": {
@@ -154,7 +232,7 @@ config = {
     }
 }
 
-# Create a proxy to multiple servers
+# Create a unified proxy to multiple servers
 composite_proxy = FastMCP.as_proxy(config, name="Composite Proxy")
 
 # Tools and resources are accessible with prefixes:
@@ -162,14 +240,72 @@ composite_proxy = FastMCP.as_proxy(config, name="Composite Proxy")
 # - weather://weather/icons/sunny, calendar://calendar/events/today
 ```
 
-## Forwarding Interactions
+## Alternative Approaches
+
+The examples above show the recommended approach using `ProxyClient` or transport strings. For advanced use cases, you can also work directly with the underlying classes.
 
-`ProxyClient` is a subclass of `Client` that implements a set of default handlers to forward advanced interactions between the backend server and the client connected to the proxy. These handlers receive requests or notifications from the backend server and forward them to the client through the related request context, relaying the response back to the backend server if needed. This setup enables the proxy to support advanced MCP features, including roots, sampling, elicitation, logging, and progress.
+### Using Regular Client
 
-To prevent the forwarding of any interaction, pass `None` to the corresponding handler when creating the `ProxyClient`. Typically, you should use `ProxyClient` to establish a proxy unless there is a specific reason not to. If the `transport` parameter is not an instance of `Client`, `FastMCP.as_proxy()` will automatically instantiate a `ProxyClient`.
+You can pass a regular `Client` instance to `as_proxy()`. The proxy will automatically create an appropriate session strategy:
+
+```python
+from fastmcp import FastMCP, Client
+
+# Using regular Client (session strategy auto-detected)
+client = Client("backend_server.py")
+proxy = FastMCP.as_proxy(client)
+```
+
+This approach provides session isolation but doesn't include advanced MCP feature forwarding (sampling, elicitation, etc.) unless you configure handlers manually.
 
 ## `FastMCPProxy` Class
 
-Internally, `FastMCP.as_proxy()` uses the `FastMCPProxy` class. You generally don't need to interact with this class directly, but it's available if needed.
+Internally, `FastMCP.as_proxy()` uses the `FastMCPProxy` class. You generally don't need to interact with this class directly, but it's available if needed for advanced scenarios.
+
+### Direct Usage
 
-Using the class directly might be necessary for advanced scenarios, like subclassing `FastMCPProxy` to add custom logic before or after forwarding requests. 
+```python
+from fastmcp.server.proxy import FastMCPProxy, ProxyClient
+
+# Provide a client factory for explicit session control
+def create_client():
+    return ProxyClient("backend_server.py")
+
+proxy = FastMCPProxy(client_factory=create_client)
+```
+
+### Parameters
+
+- **`client`**: **[DEPRECATED]** A `Client` instance. Use `client_factory` instead for explicit session management.
+- **`client_factory`**: A callable that returns a `Client` instance when called. This gives you full control over session creation and reuse strategies.
+
+### Explicit Session Management
+
+`FastMCPProxy` requires explicit session management - no automatic detection is performed. You must choose your session strategy:
+
+```python
+# Share session across all requests (be careful with concurrency)
+shared_client = ProxyClient("backend_server.py")
+def shared_session_factory():
+    return shared_client
+
+proxy = FastMCPProxy(client_factory=shared_session_factory)
+
+# Create fresh sessions per request (recommended)
+def fresh_session_factory():
+    return ProxyClient("backend_server.py")
+
+proxy = FastMCPProxy(client_factory=fresh_session_factory)
+```
+
+For automatic session strategy selection, use the convenience method `FastMCP.as_proxy()` instead.
+
+```python
+# Custom factory with specific configuration
+def custom_client_factory():
+    client = ProxyClient("backend_server.py")
+    # Add any custom configuration here
+    return client
+
+proxy = FastMCPProxy(client_factory=custom_client_factory)
+``` 

docs/servers/server.mdx

@@ -220,6 +220,8 @@ main.mount(sub, prefix="sub")
 
 FastMCP can act as a proxy for any MCP server (local or remote) using `FastMCP.as_proxy`, letting you bridge transports or add a frontend to existing servers. For example, you can expose a remote SSE server locally via stdio, or vice versa.
 
+Proxies automatically handle concurrent operations safely by creating fresh sessions for each request when using disconnected clients.
+
 See the [Proxying Servers](/servers/proxy) guide for details and advanced usage.
 
 ```python

src/fastmcp/client/client.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import datetime
 from contextlib import AsyncExitStack, asynccontextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Generic, Literal, cast, overload
+from typing import Any, Generic, Literal, TypeVar, cast, overload
 
 import anyio
 import httpx
@@ -39,6 +40,7 @@ from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.types import get_cached_typeadapter
 
 from .transports import (
+    ClientTransport,
     ClientTransportT,
     FastMCP1Server,
     FastMCPTransport,
@@ -65,6 +67,25 @@ __all__ = [
 
 logger = get_logger(__name__)
 
+T = TypeVar("T", bound="ClientTransport")
+
+
+@dataclass
+class ClientSessionState:
+    """Holds all session-related state for a Client instance.
+
+    This allows clean separation of configuration (which is copied) from
+    session state (which should be fresh for each new client instance).
+    """
+
+    session: ClientSession | None = None
+    nesting_counter: int = 0
+    lock: anyio.Lock = field(default_factory=anyio.Lock)
+    session_task: asyncio.Task | None = None
+    ready_event: anyio.Event = field(default_factory=anyio.Event)
+    stop_event: anyio.Event = field(default_factory=anyio.Event)
+    initialize_result: mcp.types.InitializeResult | None = None
+
 
 class Client(Generic[ClientTransportT]):
     """
@@ -89,8 +110,8 @@ class Client(Generic[ClientTransportT]):
     between tasks, and ensures all session state changes happen within a lock.
     Events are only created when needed, never reset outside locks.
 
-    See: https://github.com/jlowin/fastmcp/issues/1051
-         https://github.com/jlowin/fastmcp/pull/1054
+    This design prevents race conditions where tasks wait on events that get
+    replaced by other tasks, ensuring reliable coordination in concurrent scenarios.
 
     Args:
         transport:
@@ -127,56 +148,65 @@ class Client(Generic[ClientTransportT]):
     """
 
     @overload
-    def __new__(
-        cls,
-        transport: ClientTransportT,
-        **kwargs: Any,
-    ) -> Client[ClientTransportT]: ...
+    def __init__(self: Client[T], transport: T, *args, **kwargs) -> None: ...
 
     @overload
-    def __new__(
-        cls, transport: AnyUrl, **kwargs
-    ) -> Client[SSETransport | StreamableHttpTransport]: ...
+    def __init__(
+        self: Client[SSETransport | StreamableHttpTransport],
+        transport: AnyUrl,
+        *args,
+        **kwargs,
+    ) -> None: ...
 
     @overload
-    def __new__(
-        cls, transport: FastMCP | FastMCP1Server, **kwargs
-    ) -> Client[FastMCPTransport]: ...
+    def __init__(
+        self: Client[FastMCPTransport],
+        transport: FastMCP | FastMCP1Server,
+        *args,
+        **kwargs,
+    ) -> None: ...
 
     @overload
-    def __new__(
-        cls, transport: Path, **kwargs
-    ) -> Client[PythonStdioTransport | NodeStdioTransport]: ...
+    def __init__(
+        self: Client[PythonStdioTransport | NodeStdioTransport],
+        transport: Path,
+        *args,
+        **kwargs,
+    ) -> None: ...
 
     @overload
-    def __new__(
-        cls, transport: MCPConfig | dict[str, Any], **kwargs
-    ) -> Client[MCPConfigTransport]: ...
+    def __init__(
+        self: Client[MCPConfigTransport],
+        transport: MCPConfig | dict[str, Any],
+        *args,
+        **kwargs,
+    ) -> None: ...
 
     @overload
-    def __new__(
-        cls, transport: str, **kwargs
-    ) -> Client[
-        PythonStdioTransport
-        | NodeStdioTransport
-        | SSETransport
-        | StreamableHttpTransport
-    ]: ...
-
-    def __new__(cls, transport, **kwargs) -> Client:
-        instance = super().__new__(cls)
-        return instance
+    def __init__(
+        self: Client[
+            PythonStdioTransport
+            | NodeStdioTransport
+            | SSETransport
+            | StreamableHttpTransport
+        ],
+        transport: str,
+        *args,
+        **kwargs,
+    ) -> None: ...
 
     def __init__(
         self,
-        transport: ClientTransportT
-        | FastMCP
-        | AnyUrl
-        | Path
-        | MCPConfig
-        | dict[str, Any]
-        | str,
-        # Common args
+        transport: (
+            ClientTransportT
+            | FastMCP
+            | FastMCP1Server
+            | AnyUrl
+            | Path
+            | MCPConfig
+            | dict[str, Any]
+            | str
+        ),
         roots: RootsList | RootsHandler | None = None,
         sampling_handler: SamplingHandler | None = None,
         elicitation_handler: ElicitationHandler | None = None,
@@ -187,11 +217,10 @@ class Client(Generic[ClientTransportT]):
         init_timeout: datetime.timedelta | float | int | None = None,
         client_info: mcp.types.Implementation | None = None,
         auth: httpx.Auth | Literal["oauth"] | str | None = None,
-    ):
+    ) -> None:
         self.transport = cast(ClientTransportT, infer_transport(transport))
         if auth is not None:
             self.transport._set_auth(auth)
-        self._initialize_result: mcp.types.InitializeResult | None = None
 
         if log_handler is None:
             log_handler = default_log_handler
@@ -238,33 +267,26 @@ class Client(Generic[ClientTransportT]):
             )
 
         # Session context management - see class docstring for detailed explanation
-        self._session: ClientSession | None = None  # Active MCP session
-        self._nesting_counter: int = 0  # Reference count for active context managers
-        self._context_lock = anyio.Lock()  # Protects all session state changes
-        self._session_task: asyncio.Task | None = (
-            None  # Background session manager task
-        )
-        self._ready_event = anyio.Event()  # Signals when session is ready for use
-        self._stop_event = anyio.Event()  # Signals when session should stop
+        self._session_state = ClientSessionState()
 
     @property
     def session(self) -> ClientSession:
         """Get the current active session. Raises RuntimeError if not connected."""
-        if self._session is None:
+        if self._session_state.session is None:
             raise RuntimeError(
                 "Client is not connected. Use the 'async with client:' context manager first."
             )
 
-        return self._session
+        return self._session_state.session
 
     @property
     def initialize_result(self) -> mcp.types.InitializeResult:
         """Get the result of the initialization request."""
-        if self._initialize_result is None:
+        if self._session_state.initialize_result is None:
             raise RuntimeError(
                 "Client is not connected. Use the 'async with client:' context manager first."
             )
-        return self._initialize_result
+        return self._session_state.initialize_result
 
     def set_roots(self, roots: RootsList | RootsHandler) -> None:
         """Set the roots for the client. This does not automatically call `send_roots_list_changed`."""
@@ -286,7 +308,33 @@ class Client(Generic[ClientTransportT]):
 
     def is_connected(self) -> bool:
         """Check if the client is currently connected."""
-        return self._session is not None
+        return self._session_state.session is not None
+
+    def new(self) -> Client[ClientTransportT]:
+        """Create a new client instance with the same configuration but fresh session state.
+
+        This creates a new client with the same transport, handlers, and configuration,
+        but with no active session. Useful for creating independent sessions that don't
+        share state with the original client.
+
+        Returns:
+            A new Client instance with the same configuration but disconnected state.
+
+        Example:
+            ```python
+            # Create a fresh client for each concurrent operation
+            fresh_client = client.new()
+            async with fresh_client:
+                await fresh_client.call_tool("some_tool", {})
+            ```
+        """
+
+        new_client = copy.copy(self)
+
+        # Reset session state to fresh state
+        new_client._session_state = ClientSessionState()
+
+        return new_client
 
     @asynccontextmanager
     async def _context_manager(self):
@@ -294,19 +342,21 @@ class Client(Generic[ClientTransportT]):
             async with self.transport.connect_session(
                 **self._session_kwargs
             ) as session:
-                self._session = session
+                self._session_state.session = session
                 # Initialize the session
                 try:
                     with anyio.fail_after(self._init_timeout):
-                        self._initialize_result = await self._session.initialize()
+                        self._session_state.initialize_result = (
+                            await self._session_state.session.initialize()
+                        )
                     yield
                 except anyio.ClosedResourceError:
                     raise RuntimeError("Server session was closed unexpectedly")
                 except TimeoutError:
                     raise RuntimeError("Failed to initialize server session")
                 finally:
-                    self._session = None
-                    self._initialize_result = None
+                    self._session_state.session = None
+                    self._session_state.initialize_result = None
 
     async def __aenter__(self):
         return await self._connect()
@@ -328,20 +378,25 @@ class Client(Generic[ClientTransportT]):
         tasks wait on events that get replaced by other tasks.
         """
         # ensure only one session is running at a time to avoid race conditions
-        async with self._context_lock:
-            need_to_start = self._session_task is None or self._session_task.done()
+        async with self._session_state.lock:
+            need_to_start = (
+                self._session_state.session_task is None
+                or self._session_state.session_task.done()
+            )
             if need_to_start:
-                if self._nesting_counter != 0:
+                if self._session_state.nesting_counter != 0:
                     raise RuntimeError(
-                        f"Internal error: nesting counter should be 0 when starting new session, got {self._nesting_counter}"
+                        f"Internal error: nesting counter should be 0 when starting new session, got {self._session_state.nesting_counter}"
                     )
-                self._stop_event = anyio.Event()
-                self._ready_event = anyio.Event()
-                self._session_task = asyncio.create_task(self._session_runner())
-                await self._ready_event.wait()
-
-                if self._session_task.done():
-                    exception = self._session_task.exception()
+                self._session_state.stop_event = anyio.Event()
+                self._session_state.ready_event = anyio.Event()
+                self._session_state.session_task = asyncio.create_task(
+                    self._session_runner()
+                )
+                await self._session_state.ready_event.wait()
+
+                if self._session_state.session_task.done():
+                    exception = self._session_state.session_task.exception()
                     if exception is None:
                         raise RuntimeError(
                             "Session task completed without exception but connection failed"
@@ -352,7 +407,7 @@ class Client(Generic[ClientTransportT]):
                         f"Client failed to connect: {exception}"
                     ) from exception
 
-            self._nesting_counter += 1
+            self._session_state.nesting_counter += 1
         return self
 
     async def _disconnect(self, force: bool = False):
@@ -370,26 +425,28 @@ class Client(Generic[ClientTransportT]):
         Event recreation now happens only in _connect() when actually needed.
         """
         # ensure only one session is running at a time to avoid race conditions
-        async with self._context_lock:
+        async with self._session_state.lock:
             # if we are forcing a disconnect, reset the nesting counter
             if force:
-                self._nesting_counter = 0
+                self._session_state.nesting_counter = 0
 
             # otherwise decrement to check if we are done nesting
             else:
-                self._nesting_counter = max(0, self._nesting_counter - 1)
+                self._session_state.nesting_counter = max(
+                    0, self._session_state.nesting_counter - 1
+                )
 
             # if we are still nested, return
-            if self._nesting_counter > 0:
+            if self._session_state.nesting_counter > 0:
                 return
 
             # stop the active seesion
-            if self._session_task is None:
+            if self._session_state.session_task is None:
                 return
-            self._stop_event.set()
+            self._session_state.stop_event.set()
             # wait for session to finish to ensure state has been reset
-            await self._session_task
-            self._session_task = None
+            await self._session_state.session_task
+            self._session_state.session_task = None
 
     async def _session_runner(self):
         """
@@ -409,12 +466,12 @@ class Client(Generic[ClientTransportT]):
             async with AsyncExitStack() as stack:
                 await stack.enter_async_context(self._context_manager())
                 # Session/context is now ready
-                self._ready_event.set()
+                self._session_state.ready_event.set()
                 # Wait until disconnect/stop is requested
-                await self._stop_event.wait()
+                await self._session_state.stop_event.wait()
         finally:
             # Ensure ready event is set even if context manager entry fails
-            self._ready_event.set()
+            self._session_state.ready_event.set()
 
     async def close(self):
         await self._disconnect(force=True)

src/fastmcp/server/proxy.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import warnings
+from collections.abc import Callable
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, cast
 from urllib.parse import quote
@@ -16,7 +18,8 @@ from mcp.types import (
 )
 from pydantic.networks import AnyUrl
 
-from fastmcp.client import Client
+import fastmcp
+from fastmcp.client.client import Client, FastMCP1Server
 from fastmcp.client.elicitation import ElicitResult
 from fastmcp.client.logging import LogMessage
 from fastmcp.client.roots import RootsList
@@ -44,9 +47,9 @@ logger = get_logger(__name__)
 class ProxyToolManager(ToolManager):
     """A ToolManager that sources its tools from a remote client in addition to local and mounted tools."""
 
-    def __init__(self, client: Client, **kwargs):
+    def __init__(self, client_factory: Callable[[], Client], **kwargs):
         super().__init__(**kwargs)
-        self.client = client
+        self.client_factory = client_factory
 
     async def get_tools(self) -> dict[str, Tool]:
         """Gets the unfiltered tool inventory including local, mounted, and proxy tools."""
@@ -55,13 +58,12 @@ class ProxyToolManager(ToolManager):
 
         # Then add proxy tools, but don't overwrite existing ones
         try:
-            async with self.client:
-                client_tools = await self.client.list_tools()
+            client = self.client_factory()
+            async with client:
+                client_tools = await client.list_tools()
                 for tool in client_tools:
                     if tool.name not in all_tools:
-                        all_tools[tool.name] = ProxyTool.from_mcp_tool(
-                            self.client, tool
-                        )
+                        all_tools[tool.name] = ProxyTool.from_mcp_tool(client, tool)
         except McpError as e:
             if e.error.code == METHOD_NOT_FOUND:
                 pass  # No tools available from proxy
@@ -82,8 +84,9 @@ class ProxyToolManager(ToolManager):
             return await super().call_tool(key, arguments)
         except NotFoundError:
             # If not found locally, try proxy
-            async with self.client:
-                result = await self.client.call_tool(key, arguments)
+            client = self.client_factory()
+            async with client:
+                result = await client.call_tool(key, arguments)
                 return ToolResult(
                     content=result.content,
                     structured_content=result.structured_content,
@@ -93,9 +96,9 @@ class ProxyToolManager(ToolManager):
 class ProxyResourceManager(ResourceManager):
     """A ResourceManager that sources its resources from a remote client in addition to local and mounted resources."""
 
-    def __init__(self, client: Client, **kwargs):
+    def __init__(self, client_factory: Callable[[], Client], **kwargs):
         super().__init__(**kwargs)
-        self.client = client
+        self.client_factory = client_factory
 
     async def get_resources(self) -> dict[str, Resource]:
         """Gets the unfiltered resource inventory including local, mounted, and proxy resources."""
@@ -104,12 +107,13 @@ class ProxyResourceManager(ResourceManager):
 
         # Then add proxy resources, but don't overwrite existing ones
         try:
-            async with self.client:
-                client_resources = await self.client.list_resources()
+            client = self.client_factory()
+            async with client:
+                client_resources = await client.list_resources()
                 for resource in client_resources:
                     if str(resource.uri) not in all_resources:
                         all_resources[str(resource.uri)] = (
-                            ProxyResource.from_mcp_resource(self.client, resource)
+                            ProxyResource.from_mcp_resource(client, resource)
                         )
         except McpError as e:
             if e.error.code == METHOD_NOT_FOUND:
@@ -126,12 +130,13 @@ class ProxyResourceManager(ResourceManager):
 
         # Then add proxy templates, but don't overwrite existing ones
         try:
-            async with self.client:
-                client_templates = await self.client.list_resource_templates()
+            client = self.client_factory()
+            async with client:
+                client_templates = await client.list_resource_templates()
                 for template in client_templates:
                     if template.uriTemplate not in all_templates:
                         all_templates[template.uriTemplate] = (
-                            ProxyTemplate.from_mcp_template(self.client, template)
+                            ProxyTemplate.from_mcp_template(client, template)
                         )
         except McpError as e:
             if e.error.code == METHOD_NOT_FOUND:
@@ -158,8 +163,9 @@ class ProxyResourceManager(ResourceManager):
             return await super().read_resource(uri)
         except NotFoundError:
             # If not found locally, try proxy
-            async with self.client:
-                result = await self.client.read_resource(uri)
+            client = self.client_factory()
+            async with client:
+                result = await client.read_resource(uri)
                 if isinstance(result[0], TextResourceContents):
                     return result[0].text
                 elif isinstance(result[0], BlobResourceContents):
@@ -171,9 +177,9 @@ class ProxyResourceManager(ResourceManager):
 class ProxyPromptManager(PromptManager):
     """A PromptManager that sources its prompts from a remote client in addition to local and mounted prompts."""
 
-    def __init__(self, client: Client, **kwargs):
+    def __init__(self, client_factory: Callable[[], Client], **kwargs):
         super().__init__(**kwargs)
-        self.client = client
+        self.client_factory = client_factory
 
     async def get_prompts(self) -> dict[str, Prompt]:
         """Gets the unfiltered prompt inventory including local, mounted, and proxy prompts."""
@@ -182,12 +188,13 @@ class ProxyPromptManager(PromptManager):
 
         # Then add proxy prompts, but don't overwrite existing ones
         try:
-            async with self.client:
-                client_prompts = await self.client.list_prompts()
+            client = self.client_factory()
+            async with client:
+                client_prompts = await client.list_prompts()
                 for prompt in client_prompts:
                     if prompt.name not in all_prompts:
                         all_prompts[prompt.name] = ProxyPrompt.from_mcp_prompt(
-                            self.client, prompt
+                            client, prompt
                         )
         except McpError as e:
             if e.error.code == METHOD_NOT_FOUND:
@@ -213,8 +220,9 @@ class ProxyPromptManager(PromptManager):
             return await super().render_prompt(name, arguments)
         except NotFoundError:
             # If not found locally, try proxy
-            async with self.client:
-                result = await self.client.get_prompt(name, arguments)
+            client = self.client_factory()
+            async with client:
+                result = await client.get_prompt(name, arguments)
                 return result
 
 
@@ -245,7 +253,6 @@ class ProxyTool(Tool):
         context: Context | None = None,
     ) -> ToolResult:
         """Executes the tool by making a call through the client."""
-        # This is where the remote execution logic lives.
         async with self._client:
             result = await self._client.call_tool_mcp(
                 name=self.name,
@@ -267,14 +274,22 @@ class ProxyResource(Resource):
     _client: Client
     _value: str | bytes | None = None
 
-    def __init__(self, client: Client, *, _value: str | bytes | None = None, **kwargs):
+    def __init__(
+        self,
+        client: Client,
+        *,
+        _value: str | bytes | None = None,
+        **kwargs,
+    ):
         super().__init__(**kwargs)
         self._client = client
         self._value = _value
 
     @classmethod
     def from_mcp_resource(
-        cls, client: Client, mcp_resource: mcp.types.Resource
+        cls,
+        client: Client,
+        mcp_resource: mcp.types.Resource,
     ) -> ProxyResource:
         """Factory method to create a ProxyResource from a raw MCP resource schema."""
         return cls(
@@ -397,24 +412,63 @@ class ProxyPrompt(Prompt):
 class FastMCPProxy(FastMCP):
     """
     A FastMCP server that acts as a proxy to a remote MCP-compliant server.
-    It uses specialized managers that fulfill requests via an HTTP client.
+    It uses specialized managers that fulfill requests via a client factory.
     """
 
-    def __init__(self, client: Client, **kwargs):
+    def __init__(
+        self,
+        client: Client | None = None,
+        *,
+        client_factory: Callable[[], Client] | None = None,
+        **kwargs,
+    ):
         """
         Initializes the proxy server.
 
+        FastMCPProxy requires explicit session management via client_factory.
+        Use FastMCP.as_proxy() for convenience with automatic session strategy.
+
         Args:
-            client: The FastMCP client connected to the backend server.
+            client: [DEPRECATED] A Client instance. Use client_factory instead for explicit
+                   session management. When provided, a client_factory will be automatically
+                   created that provides session isolation for backwards compatibility.
+            client_factory: A callable that returns a Client instance when called.
+                           This gives you full control over session creation and reuse.
             **kwargs: Additional settings for the FastMCP server.
         """
+
         super().__init__(**kwargs)
-        self.client = client
+
+        # Handle client and client_factory parameters
+        if client is not None and client_factory is not None:
+            raise ValueError("Cannot specify both 'client' and 'client_factory'")
+
+        if client is not None:
+            # Deprecated in 2.10.3
+            if fastmcp.settings.deprecation_warnings:
+                warnings.warn(
+                    "Passing 'client' to FastMCPProxy is deprecated. Use 'client_factory' instead for explicit session management. "
+                    "For automatic session strategy, use FastMCP.as_proxy().",
+                    DeprecationWarning,
+                    stacklevel=2,
+                )
+
+            # Create a factory that provides session isolation for backwards compatibility
+            def deprecated_client_factory():
+                return client.new()
+
+            self.client_factory = deprecated_client_factory
+        elif client_factory is not None:
+            self.client_factory = client_factory
+        else:
+            raise ValueError("Must specify 'client_factory'")
 
         # Replace the default managers with our specialized proxy managers.
-        self._tool_manager = ProxyToolManager(client=self.client)
-        self._resource_manager = ProxyResourceManager(client=self.client)
-        self._prompt_manager = ProxyPromptManager(client=self.client)
+        self._tool_manager = ProxyToolManager(client_factory=self.client_factory)
+        self._resource_manager = ProxyResourceManager(
+            client_factory=self.client_factory
+        )
+        self._prompt_manager = ProxyPromptManager(client_factory=self.client_factory)
 
 
 async def default_proxy_roots_handler(
@@ -435,15 +489,14 @@ class ProxyClient(Client[ClientTransportT]):
 
     def __init__(
         self,
-        transport: (
-            ClientTransportT
-            | FastMCP
-            | AnyUrl
-            | Path
-            | MCPConfig
-            | dict[str, Any]
-            | str
-        ),
+        transport: ClientTransportT
+        | FastMCP
+        | FastMCP1Server
+        | AnyUrl
+        | Path
+        | MCPConfig
+        | dict[str, Any]
+        | str,
         **kwargs,
     ):
         if "roots" not in kwargs:
@@ -456,7 +509,7 @@ class ProxyClient(Client[ClientTransportT]):
             kwargs["log_handler"] = ProxyClient.default_log_handler
         if "progress_handler" not in kwargs:
             kwargs["progress_handler"] = ProxyClient.default_progress_handler
-        super().__init__(transport, **kwargs)
+        super().__init__(**kwargs | dict(transport=transport))
 
     @classmethod
     async def default_sampling_handler(

src/fastmcp/server/server.py

@@ -1931,10 +1931,39 @@ class FastMCP(Generic[LifespanResultT]):
 
         if isinstance(backend, Client):
             client = backend
+            # Session strategy based on client connection state:
+            # - Connected clients: reuse existing session for all requests
+            # - Disconnected clients: create fresh sessions per request for isolation
+            if client.is_connected():
+                from fastmcp.utilities.logging import get_logger
+
+                logger = get_logger(__name__)
+                logger.info(
+                    "Proxy detected connected client - reusing existing session for all requests. "
+                    "This may cause context mixing in concurrent scenarios."
+                )
+
+                # Reuse sessions - return the same client instance
+                def reuse_client_factory():
+                    return client
+
+                client_factory = reuse_client_factory
+            else:
+                # Fresh sessions per request
+                def fresh_client_factory():
+                    return client.new()
+
+                client_factory = fresh_client_factory
         else:
-            client = ProxyClient(backend)
+            base_client = ProxyClient(backend)
+
+            # Fresh client created from transport - use fresh sessions per request
+            def proxy_client_factory():
+                return base_client.new()
+
+            client_factory = proxy_client_factory
 
-        return FastMCPProxy(client=client, **settings)
+        return FastMCPProxy(client_factory=client_factory, **settings)
 
     @classmethod
     def from_client(

tests/client/test_client.py

@@ -449,27 +449,27 @@ async def test_client_nested_context_manager(fastmcp_server):
 
     # Before connection
     assert not client.is_connected()
-    assert client._session is None
+    assert client._session_state.session is None
 
     # During connection
     async with client:
         assert client.is_connected()
-        assert client._session is not None
-        session = client._session
+        assert client._session_state.session is not None
+        session = client._session_state.session
 
         # Re-use the same session
         async with client:
             assert client.is_connected()
-            assert client._session is session
+            assert client._session_state.session is session
 
         # Re-use the same session
         async with client:
             assert client.is_connected()
-            assert client._session is session
+            assert client._session_state.session is session
 
     # After connection
     assert not client.is_connected()
-    assert client._session is None
+    assert client._session_state.session is None
 
 
 async def test_concurrent_client_context_managers():

tests/deprecated/test_proxy_client.py

@@ -0,0 +1,107 @@
+"""Tests for deprecated FastMCPProxy client parameter."""
+
+import warnings
+
+import pytest
+
+from fastmcp import Client, FastMCP
+from fastmcp.server.proxy import FastMCPProxy, ProxyClient
+
+
+@pytest.fixture
+def simple_server():
+    """Create a simple FastMCP server for testing."""
+    server = FastMCP("TestServer")
+
+    @server.tool
+    def simple_tool() -> str:
+        return "test_result"
+
+    return server
+
+
+class TestDeprecatedClientParameter:
+    """Test the deprecated client parameter in FastMCPProxy."""
+
+    def test_client_parameter_deprecation_warning(self, simple_server):
+        """Test that using the client parameter raises a deprecation warning."""
+        client = Client(simple_server)
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")  # Ensure all warnings are captured
+
+            FastMCPProxy(client=client)
+
+            # Verify a deprecation warning was raised
+            assert len(w) == 1
+            assert issubclass(w[0].category, DeprecationWarning)
+            assert "client' to FastMCPProxy is deprecated" in str(w[0].message)
+            assert "client_factory" in str(w[0].message)
+
+    def test_client_parameter_still_works(self, simple_server):
+        """Test that the deprecated client parameter still functions."""
+        client = ProxyClient(simple_server)
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")  # Suppress warnings for functionality test
+
+            proxy = FastMCPProxy(client=client)
+
+            # Verify the proxy was created successfully
+            assert proxy is not None
+            assert hasattr(proxy, "client_factory")
+            assert callable(proxy.client_factory)
+
+            # Verify the factory returns a new client instance (session isolation for backwards compatibility)
+            returned_client = proxy.client_factory()
+            assert returned_client is not client
+            assert isinstance(returned_client, type(client))
+
+    def test_cannot_specify_both_client_and_factory(self, simple_server):
+        """Test that specifying both client and client_factory raises an error."""
+        client = Client(simple_server)
+
+        def factory():
+            return Client(simple_server)
+
+        with pytest.raises(
+            ValueError, match="Cannot specify both 'client' and 'client_factory'"
+        ):
+            FastMCPProxy(client=client, client_factory=factory)
+
+    def test_must_specify_client_factory_when_no_client(self):
+        """Test that client_factory is required when client is not provided."""
+        with pytest.raises(ValueError, match="Must specify 'client_factory'"):
+            FastMCPProxy()
+
+    def test_client_factory_preferred_over_deprecated_client(self, simple_server):
+        """Test that the recommended client_factory approach works without warnings."""
+
+        def factory():
+            return ProxyClient(simple_server)
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+
+            proxy = FastMCPProxy(client_factory=factory)
+
+            # Verify no warnings were raised
+            assert len(w) == 0
+
+            # Verify the proxy works correctly
+            assert proxy is not None
+            assert proxy.client_factory is factory
+
+    async def test_deprecated_client_functional_test(self, simple_server):
+        """End-to-end test that deprecated client parameter still works functionally."""
+        client = ProxyClient(simple_server)
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore")
+
+            proxy = FastMCPProxy(client=client)
+
+        # Test that the proxy can actually handle requests
+        async with Client(proxy) as proxy_client:
+            result = await proxy_client.call_tool("simple_tool", {})
+            assert result.data == "test_result"

tests/server/proxy/test_proxy_client.py

@@ -2,6 +2,7 @@ from dataclasses import dataclass
 from typing import cast
 
 import pytest
+from anyio import create_task_group
 from mcp.types import LoggingLevel, ModelHint, ModelPreferences, TextContent
 
 from fastmcp import Client, Context, FastMCP
@@ -255,3 +256,104 @@ class TestProxyClient:
             await client.call_tool("report_progress", {})
 
         assert PROGRESS_MESSAGES == EXPECTED_PROGRESS_MESSAGES
+
+    async def test_concurrent_log_requests_no_mixing(self, proxy_server: FastMCP):
+        """Test that concurrent log requests don't mix handlers (fixes #1068)."""
+        results: dict[str, LogMessage] = {}
+
+        async def log_handler_a(message: LogMessage) -> None:
+            results["logger_a"] = message
+
+        async def log_handler_b(message: LogMessage) -> None:
+            results["logger_b"] = message
+
+        async with (
+            Client(proxy_server, log_handler=log_handler_a) as client_a,
+            Client(proxy_server, log_handler=log_handler_b) as client_b,
+        ):
+            async with create_task_group() as tg:
+                tg.start_soon(
+                    client_a.call_tool,
+                    "log",
+                    {"message": "Hello, world!", "level": "info", "logger": "a"},
+                )
+                tg.start_soon(
+                    client_b.call_tool,
+                    "log",
+                    {"message": "Hello, world!", "level": "info", "logger": "b"},
+                )
+
+        assert results["logger_a"].logger == "a"
+        assert results["logger_b"].logger == "b"
+
+    async def test_concurrent_elicitation_no_mixing(self, proxy_server: FastMCP):
+        """Test that concurrent elicitation requests don't mix handlers (fixes #1068)."""
+        results = {}
+
+        async def elicitation_handler_a(
+            message: str,
+            response_type: type,
+            params: ElicitRequestParams,
+            ctx: RequestContext,
+        ) -> ElicitResult:
+            return ElicitResult(action="accept", content=response_type(name="Alice"))
+
+        async def elicitation_handler_b(
+            message: str,
+            response_type: type,
+            params: ElicitRequestParams,
+            ctx: RequestContext,
+        ) -> ElicitResult:
+            return ElicitResult(action="accept", content=response_type(name="Bob"))
+
+        async def get_and_store(name, coro):
+            result = await coro
+            results[name] = result.data
+
+        async with (
+            Client(proxy_server, elicitation_handler=elicitation_handler_a) as client_a,
+            Client(proxy_server, elicitation_handler=elicitation_handler_b) as client_b,
+        ):
+            async with create_task_group() as tg:
+                tg.start_soon(
+                    get_and_store,
+                    "elicitation_a",
+                    client_a.call_tool("elicit", {}),
+                )
+                tg.start_soon(
+                    get_and_store,
+                    "elicitation_b",
+                    client_b.call_tool("elicit", {}),
+                )
+
+        assert results["elicitation_a"] == "Hello, Alice!"
+        assert results["elicitation_b"] == "Hello, Bob!"
+
+    async def test_client_factory_creates_fresh_sessions(self, fastmcp_server: FastMCP):
+        """Test that the client factory pattern creates fresh sessions for each request."""
+        from fastmcp.server.proxy import FastMCPProxy
+
+        # Create a disconnected client (should use fresh sessions per request)
+        base_client = Client(fastmcp_server)
+
+        # Test both as_proxy convenience method and direct client_factory usage
+        proxy_via_as_proxy = FastMCP.as_proxy(base_client)
+        proxy_via_factory = FastMCPProxy(client_factory=base_client.new)
+
+        # Verify the proxies are created successfully - this tests the client factory pattern
+        assert proxy_via_as_proxy is not None
+        assert proxy_via_factory is not None
+
+        # Verify they have the expected client factory behavior
+        assert hasattr(proxy_via_as_proxy, "_tool_manager")
+        assert hasattr(proxy_via_factory, "_tool_manager")
+
+    async def test_connected_client_reuses_sessions(self, fastmcp_server: FastMCP):
+        """Test that connected clients passed to as_proxy reuse sessions (preserves #959 behavior)."""
+        # Create a connected client (should reuse sessions)
+        async with Client(fastmcp_server) as connected_client:
+            proxy = FastMCP.as_proxy(connected_client)
+
+            # Verify the proxy is created successfully and uses session reuse
+            assert proxy is not None
+            assert hasattr(proxy, "_tool_manager")

tests/server/{test_proxy.py → proxy/test_proxy_server.py}

@@ -106,8 +106,8 @@ def test_as_proxy_with_url():
     """FastMCP.as_proxy should accept a URL without connecting."""
     proxy = FastMCP.as_proxy("http://example.com/mcp/")
     assert isinstance(proxy, FastMCPProxy)
-    assert isinstance(proxy.client.transport, StreamableHttpTransport)
-    assert proxy.client.transport.url == "http://example.com/mcp/"
+    assert isinstance(proxy.client_factory().transport, StreamableHttpTransport)
+    assert proxy.client_factory().transport.url == "http://example.com/mcp/"  # type: ignore[attr-defined]
 
 
 class TestTools: