Can someone provide an example of sampling feature?

[shrap42]

I am trying to use sampling as part of my workflow. The workflow is kinda complicated but the simple example where I would have an fast agent client connected to MCP server (via HTTP streamable) - this I have working - that would call a tool that would call back the client with a sampling request (so far I have it working, I can see the incoming request in the debug log), then the client would call the LLM, probably with the Human in the loop accepting the prompt, then the response being sent back to the server tool and finally response from the tool would be sent back to the original client.

I thought all I need to do is:

1. have a tool similar to this (I change translate to custom transcode to avoid direct translation by llm)

@mcp.tool()
async def custom_transcode(context:Context, text:str, target_lang: str) -> str:
    prompt = f"translate the following text to {target_lang}: {text}"
    translation = await context.sample(prompt)
    return translation.text

2. have MCP server set in a way that the sampling is offered by the client

mcp:
    servers:
        mcphttp:
            transport: "http"
            url: "http://localhost:8008/mcp/"   
            timeout: 30    
            sampling:
                model: "openai"   

And that's it. Am I missing something? It's not working, though. The agent looks like this:

import asyncio
import logging
import sys

from mcp_agent.core.fastagent import FastAgent

# Set up logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("agent")

# Create the application
fast = FastAgent("fast-agent example")

# Define the agent
@fast.agent(
    instruction="You are a helpful AI Agent that can add numbers",
    servers=["mcphttp"],
    human_input=True
)


async def main():
    try:
        logger.info("Starting agent run...")
        async with fast.run() as agent:
            logger.info("Agent initialized")
            await agent.interactive()            

    except Exception as e:
        logger.error(f"Error initializing agent: {e}", exc_info=True)
        sys.exit(1)

if __name__ == "__main__":
    asyncio.run(main())

and it hangs when it receives the sampling request.

I tried the same without the interactive() method, just directly send the message to the agent and the result was same. Any suggestions?

[shrap42]

@evalstate I've noticed there was an issue in mcp regarding this? You discussed it in #187, right? So the new release should fix it for me?

[evalstate]

Yep, I'll do a release tonight - the branch on fix/misc definitely works as I've been testing Sampling and Streaming HTTP with it (and was actually unaware of this specific issue!). Sampling is in the integration test suite, but only as a STDIO server. The MCP API is 1.9.1 on that branch (so you will need to uv sync to get going).

[shrap42]

Ok, the latest release helped, it works. Now, to my real use case 🙂
I would like to call sampling on client 1 (that would be fast-agent client) while processing tool call from client 2. I am using FastMCP as a server. Is that possible? Or is that possible if I use fast-agent as MCP server?

[evalstate]

So, you have an MCP Server with 2 different clients connected? My latest version of mcp-webcam makes me thing the technique is definitely possible, but it's typescript - don't know if the FastMCP architecture would be a blocker for this. But as long as you know which session you are after - sure!