MCP Sampling #1569

Activity

1. Gerr1ck

   added

   questionFurther information is requestedFurther information is requested

   on Nov 4, 2025
2. maxisbey commented on Nov 4, 2025

   

   maxisbey

   on Nov 4, 2025

   Contributor

   More actions

   1. Yes the MCP server can initiate a sampling request whenever it wants without the client having first sent a tool call. Although, this does require a stateful connection to be open, whether that be the now deprecated SSE transport, or Streamable HTTP in stateful mode.
   2. We're currently pretty lacking on docs for sampling, see here for a recent PR adding examples: [DOC] simple mcp client with sampling capability #1436
3. Gerr1ck commented on Nov 4, 2025

   

   Gerr1ck

   on Nov 4, 2025

   Author

   More actions

   @maxisbey okay understood. I need to use a stateful connection like Streamable HTTP. Before, I was trying to implement this Use Case with STDIO...

   Regarding to the second bulletpoint: This only covers a MCP Client which enables sampling capability, as far I understood. Thats only one piece.

   The other interesting part would be how a MCP Server sends a sampling message to the client. How can I implement this? Any hints which methods from the SDK shall I use for this?
4. yarnabrina commented on Nov 4, 2025

   

   yarnabrina

   on Nov 4, 2025

   Last edited by yarnabrina

   More actions

   Hi @Gerr1ck , I'm not familiar with any official docs but if a dummy sample is enough for you to get a first look (this is what I tested the PR client initially with), you can take a look here.

   https://github.com/yarnabrina/learn-model-context-protocol/blob/main/src/mcp_learning/mcp_server/simplification.py#L13-L66

   (This is not stateful though.)
5. Gerr1ck commented on Nov 5, 2025

   

   Gerr1ck

   on Nov 5, 2025

   Author

   More actions

   Hi @yarnabrina thank you the link.

   Question to that:
   I cannot find a reference where the method
   async def parse_arithmetic_expression(text: str, context: Context) -> str:
   gets called.

   Can you explain me how this method gets called and which context shall be used?

   Imagine that the MCP Server just want to call parse_arithmetic_expression itself. From where do the MCP Server gets the context?
6. yarnabrina commented on Nov 5, 2025

   

   yarnabrina

   on Nov 5, 2025

   More actions

   There is no explicit call to parse_arithmetic_expression. It's an MCP tool, and converted and exposed that way here. I could have used .tool decorator as well, would have worked similarly.

   It never needs to be called explicitly, based on the available tools and contracts, LLM decides what to call with what argument and then MCP client calls the same. You can use my repo above for client example, or the PR I created also have a minimal version.

   I'm not exactly familiar with context creation, I believe it's set at low level, and it's part of the contract if you want to use sampling/elicitation. But hopefully @maxisbey or @felixweinberger can answer in details more, I'm not too familiar at the low level details.
7. maxisbey commented on Mar 24, 2026

   

   maxisbey

   on Mar 24, 2026

   Contributor

   More actions

   Hi @Gerr1ck, to answer your follow-up directly:

   Imagine that the MCP Server just wants to call [sampling] itself. From where does the MCP Server get the context?

   There are two scenarios here:

   1. Sampling from inside a tool (the common case)

   The Context is injected automatically — you never construct it yourself. Just add a Context-typed parameter to your tool and the framework fills it in when the client calls the tool. Then use ctx.session.create_message():

   Server — FastMCP (v1.x) / MCPServer (main)

   from mcp.server.fastmcp import Context, FastMCP  # on main: from mcp.server.mcpserver import Context, MCPServer
   from mcp.types import SamplingMessage, TextContent
   
   mcp = FastMCP("sampling-demo")
   
   
   @mcp.tool()
   async def summarize(text: str, ctx: Context) -> str:
       """Ask the client's LLM to summarize some text."""
       result = await ctx.session.create_message(
           messages=[
               SamplingMessage(
                   role="user",
                   content=TextContent(type="text", text=f"Summarize: {text}"),
               )
           ],
           max_tokens=100,
       )
       return result.content.text if result.content.type == "text" else str(result.content)
   
   
   if __name__ == "__main__":
       mcp.run()

   Client

   import anyio
   from mcp.client.session import ClientSession
   from mcp.client.stdio import StdioServerParameters, stdio_client
   from mcp.types import CreateMessageResult, TextContent
   
   
   async def handle_sampling(context, params) -> CreateMessageResult:
       # In a real client you'd call your LLM here
       return CreateMessageResult(
           role="assistant",
           content=TextContent(type="text", text="A brief summary."),
           model="mock-model",
           stopReason="endTurn",  # stop_reason= on main
       )
   
   
   async def main():
       params = StdioServerParameters(command="python", args=["server.py"])
       async with stdio_client(params) as (read, write):
           async with ClientSession(read, write, sampling_callback=handle_sampling) as session:
               await session.initialize()
               result = await session.call_tool("summarize", {"text": "long document..."})
               print(result.content[0].text)
   
   
   anyio.run(main)

   2. Server-initiated sampling (no tool call)

   If you want the server to sample on its own schedule without the client calling a tool first, Context won't help — it only exists inside request handlers. You need to drop to the low-level ServerSession directly.

   The catch: server.run() creates the ServerSession internally and doesn't expose it. The workaround is to skip server.run(), create the session yourself, and run the message loop alongside your own background task.

   This works over stdio (it's a stateful connection, no need for HTTP):

   Server

   import sys
   import anyio
   from mcp.server import Server
   from mcp.server.session import InitializationState, ServerSession
   from mcp.server.stdio import stdio_server
   from mcp.types import SamplingMessage, TextContent
   
   server = Server("sampling-demo")
   
   
   async def background_work(session: ServerSession) -> None:
       # Wait for the client to finish the initialize handshake
       while session._initialization_state != InitializationState.Initialized:
           await anyio.sleep(0.05)
   
       # Now the server can sample whenever it wants — no tool call needed
       for i in range(3):
           result = await session.create_message(
               messages=[
                   SamplingMessage(
                       role="user",
                       content=TextContent(type="text", text=f"Say hello #{i}"),
                   )
               ],
               max_tokens=100,
           )
           print(f"[server] got: {result.content.text}", file=sys.stderr)
           await anyio.sleep(0.5)
   
   
   async def main() -> None:
       async with stdio_server() as (read, write):
           # Create the session ourselves instead of calling server.run()
           async with ServerSession(read, write, server.create_initialization_options()) as session:
               async with anyio.create_task_group() as tg:
                   tg.start_soon(background_work, session)
                   # The normal message-processing loop
                   async for msg in session.incoming_messages:
                       tg.start_soon(server._handle_message, msg, session, {}, False)
   
   
   anyio.run(main)

   Client

   import anyio
   from mcp.client.session import ClientSession
   from mcp.client.stdio import StdioServerParameters, stdio_client
   from mcp.types import CreateMessageRequestParams, CreateMessageResult, TextContent
   
   
   async def handle_sampling(context, params: CreateMessageRequestParams) -> CreateMessageResult:
       user_text = params.messages[0].content.text
       print(f"[client] server asked: {user_text!r}")
       # In a real client you'd call your LLM here
       return CreateMessageResult(
           role="assistant",
           content=TextContent(type="text", text=f"(mock reply to: {user_text})"),
           model="mock-model",
           stopReason="endTurn",  # stop_reason= on main
       )
   
   
   async def main() -> None:
       params = StdioServerParameters(command="python", args=["server.py"])
       async with stdio_client(params) as (read, write):
           async with ClientSession(read, write, sampling_callback=handle_sampling) as session:
               await session.initialize()
               # Keep the connection open so the server's background task can sample
               await anyio.sleep(3)
   
   
   anyio.run(main)

   Output

   $ python client.py
   [client] server asked: 'Say hello #0'
   [server] got: (mock reply to: Say hello #0)
   [client] server asked: 'Say hello #1'
   [server] got: (mock reply to: Say hello #1)
   [client] server asked: 'Say hello #2'
   [server] got: (mock reply to: Say hello #2)
   

   Notes

   • Approach 2 pokes at two private-ish members (_initialization_state and _handle_message). There isn't currently a cleaner public hook for "give me the session once a client connects" — if that's something you'd find useful as a first-class API, feel free to open a feature request.
   • Both approaches work identically on v1.x and main. The only differences are import paths (FastMCP → MCPServer) and field naming (stopReason → stop_reason).
   • For Streamable HTTP in stateful mode the idea is the same, but you'd capture the session from StreamableHTTPSessionManager._server_instances instead (one session per connected client).

   Closing this for now — feel free to reopen if you have further questions.

   _{AI Disclaimer}
8. maxisbey

   closed this as completedon Mar 24, 2026