Unify SamplingHandler and promote OpenAI handler

docs/clients/sampling.mdx

@@ -185,7 +185,7 @@ For full-featured sampling with tool support, use the built-in OpenAI handler. I
 
 ```python
 from fastmcp import Client
-from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
+from fastmcp.client.sampling.handlers.openai import OpenAISamplingHandler
 
 client = Client(
     "my_mcp_server.py",
@@ -212,5 +212,5 @@ Tool execution happens on the server side. The client's role is to pass tools to
 </Note>
 
 <Tip>
-To implement a custom sampling handler, see the [OpenAISamplingHandler source code](https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/experimental/sampling/handlers/openai.py) as a reference.
+To implement a custom sampling handler, see the [OpenAISamplingHandler source code](https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/client/sampling/handlers/openai.py) as a reference.
 </Tip>

docs/servers/sampling.mdx

@@ -461,9 +461,7 @@ FastMCP provides an OpenAI-compatible handler that works with OpenAI's API and c
 import os
 from openai import OpenAI
 from fastmcp import FastMCP
-from fastmcp.experimental.sampling.handlers.openai import (
-    OpenAISamplingHandler,
-)
+from fastmcp.client.sampling.handlers.openai import OpenAISamplingHandler
 
 server = FastMCP(
     name="My Server",

examples/advanced_sampling/client_sampling_test.py

@@ -21,7 +21,7 @@
 from pydantic import BaseModel
 
 from fastmcp import Client, Context, FastMCP
-from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
+from fastmcp.client.sampling.handlers.openai import OpenAISamplingHandler
 
 # Create the MCP server
 mcp = FastMCP("Sampling Test Server")

examples/advanced_sampling/structured_output.py

@@ -18,7 +18,7 @@
 from pydantic import BaseModel
 
 from fastmcp import Client, Context, FastMCP
-from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
+from fastmcp.client.sampling.handlers.openai import OpenAISamplingHandler
 
 
 # Define a structured output model

examples/advanced_sampling/tool_use.py

@@ -18,7 +18,7 @@
 from pydantic import BaseModel, Field
 
 from fastmcp import Client, Context, FastMCP
-from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
+from fastmcp.client.sampling.handlers.openai import OpenAISamplingHandler
 
 
 # Define tools (available to the LLM during sampling)

examples/sampling_fallback.py

@@ -9,7 +9,7 @@
 from openai import OpenAI
 
 from fastmcp import FastMCP
-from fastmcp.experimental.sampling.handlers.openai import OpenAISamplingHandler
+from fastmcp.client.sampling.handlers.openai import OpenAISamplingHandler
 from fastmcp.server.context import Context
 
 

src/fastmcp/client/client.py

@@ -47,7 +47,6 @@
     create_roots_callback,
 )
 from fastmcp.client.sampling import (
-    ClientSamplingHandler,
     SamplingHandler,
     create_sampling_callback,
 )
@@ -82,7 +81,6 @@
 
 __all__ = [
     "Client",
-    "ClientSamplingHandler",
     "ElicitationHandler",
     "LogHandler",
     "MessageHandler",
@@ -248,7 +246,7 @@ def __init__(
         ),
         name: str | None = None,
         roots: RootsList | RootsHandler | None = None,
-        sampling_handler: ClientSamplingHandler | None = None,
+        sampling_handler: SamplingHandler | None = None,
         sampling_capabilities: mcp.types.SamplingCapability | None = None,
         elicitation_handler: ElicitationHandler | None = None,
         log_handler: LogHandler | None = None,
@@ -368,7 +366,7 @@ def set_roots(self, roots: RootsList | RootsHandler) -> None:
 
     def set_sampling_callback(
         self,
-        sampling_callback: ClientSamplingHandler,
+        sampling_callback: SamplingHandler,
         sampling_capabilities: mcp.types.SamplingCapability | None = None,
     ) -> None:
         """Set the sampling callback for the client."""

src/fastmcp/client/sampling/__init__.py

@@ -0,0 +1,69 @@
+import inspect
+from collections.abc import Awaitable, Callable
+from typing import TypeAlias, TypeVar
+
+import mcp.types
+from mcp import ClientSession, CreateMessageResult
+from mcp.client.session import SamplingFnT
+from mcp.server.session import ServerSession
+from mcp.shared.context import LifespanContextT, RequestContext
+from mcp.types import CreateMessageRequestParams as SamplingParams
+from mcp.types import CreateMessageResultWithTools, SamplingMessage
+
+# Result type that handlers can return
+SamplingHandlerResult: TypeAlias = (
+    str | CreateMessageResult | CreateMessageResultWithTools
+)
+
+# Session type for sampling handlers - works with both client and server sessions
+SessionT = TypeVar("SessionT", ClientSession, ServerSession)
+
+# Unified sampling handler type that works for both clients and servers.
+# Handlers receive messages and parameters from the MCP sampling flow
+# and return LLM responses.
+SamplingHandler: TypeAlias = Callable[
+    [
+        list[SamplingMessage],
+        SamplingParams,
+        RequestContext[SessionT, LifespanContextT],
+    ],
+    SamplingHandlerResult | Awaitable[SamplingHandlerResult],
+]
+
+
+__all__ = [
+    "RequestContext",
+    "SamplingHandler",
+    "SamplingHandlerResult",
+    "SamplingMessage",
+    "SamplingParams",
+    "create_sampling_callback",
+]
+
+
+def create_sampling_callback(
+    sampling_handler: SamplingHandler,
+) -> SamplingFnT:
+    async def _sampling_handler(
+        context,
+        params: SamplingParams,
+    ) -> CreateMessageResult | CreateMessageResultWithTools | mcp.types.ErrorData:
+        try:
+            result = sampling_handler(params.messages, params, context)
+            if inspect.isawaitable(result):
+                result = await result
+
+            if isinstance(result, str):
+                result = CreateMessageResult(
+                    role="assistant",
+                    model="fastmcp-client",
+                    content=mcp.types.TextContent(type="text", text=result),
+                )
+            return result
+        except Exception as e:
+            return mcp.types.ErrorData(
+                code=mcp.types.INTERNAL_ERROR,
+                message=str(e),
+            )
+
+    return _sampling_handler