工具可以通过采样（生成文本）与大型语言模型（LLMs）进行交互。

采样sampling演示：mcp服务端通过上下文来回调客户端的大模型，得到结果。

服务端代码：

"""6.5 【上下文类】Context 大模型sampling服务端开发"""
from mcp import SamplingMessage
from mcp.server.fastmcp import FastMCP,Context
from mcp.types import TextContent

#创建mcp实例
mcp = FastMCP("Context with sampling MCP Server")

@mcp.tool()
async def sampling_tool(ctx:Context):
    """
    大模型sampling服务端开发
    :param ctx: 上下文类
    :return: 处理完成
    """
    response = await ctx.session.create_message(
        messages=[
          SamplingMessage(
              role="user",
              content = TextContent(
                  text="帮我编写一个小学放假通知的文章，字数在100-200字之间。",
                  type="text",
              )
          )
        ],max_tokens=1024
    )
    #打印response
    print("response:",response)
    return "大模型sampling回调完成"



if __name__ == '__main__':
    print("MCP Server is running...")
    mcp.run(transport='sse')

客户端代码：

import asyncio
import json
from contextlib import AsyncExitStack
from typing import Optional
from mcp import ClientSession
from mcp.server.fastmcp import Context
from mcp.types import CreateMessageRequestParams, CreateMessageResult, TextContent
from openai import OpenAI
from mcp.client.sse import sse_client


class MCPClient:
    def __init__(self):
        # 创建线程管理栈
        self.exit_stack = AsyncExitStack()
        self.session: Optional[ClientSession] = None
        self.ali = OpenAI(
            api_key="sk-3ad75cfac**********f8d3d473b2a",
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
        )

    async def sampling_handler(self, context: Context, params: CreateMessageRequestParams) -> CreateMessageResult:
        print(f"context: {context}")
        print(f"params: {params}")
        messages = [
            {
                "role": message.role,
                "content": message.content.text
            } for message in params.messages
        ]
        # 调用大模型
        model_response = self.ali.chat.completions.create(
            model="qwen-turbo",
            messages=messages,
        )
        # 获取大模型的决策结果
        print("==== deepseek 决策结果：", model_response)
        choice_result = model_response.choices[0]
        return CreateMessageResult(
            role="assistant",
            content=TextContent(text=choice_result.message.content,
                                type="text"),
            model="deepseek-chat"
        )


    """创建连接服务端"""


    async def connect_to_server(self, server_path: str):
        # 创建sse_client
        client = sse_client(url=server_path)
        stdio_transport = await self.exit_stack.enter_async_context(client)
        read_stream, write_stream = stdio_transport
        # 三、创建ClientSession
        client_session = ClientSession(
            read_stream,
            write_stream,
            sampling_callback=self.sampling_handler,
        )
        self.session = await self.exit_stack.enter_async_context(client_session)
        # 四、初始化session
        await self.session.initialize()
        # 五、列出所有可用的工具
        response = await self.session.list_tools()
        tools = response.tools
        tool = tools[0]
        print("\nConnected to server with tools:", tools)
        result = await self.session.call_tool(tool.name)
        print("-----------调用工具返回结果：", result)


    # 关闭连接
    async def cleanup(self):
        await self.exit_stack.aclose()


async def main():
    client = MCPClient()
    try:
        await client.connect_to_server("http://localhost:8000/sse")
    except Exception as e:
        print(f"Error: {str(e)}")
    finally:
        await client.cleanup()


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

这里是用了阿里百炼平台注册的api_key。

如果用deepseek，配置改动如下；

self.deepseek = OpenAI(
            api_key="sk-56ad7220d4******d6680dca495000",
            base_url="https://api.deepseek.com"
        )

# 调用大模型
        deepseek_response = self.deepseek.chat.completions.create(
            model="deepseek-chat",
            messages=messages,
        )

运行结果如下：

Connected to server with tools: [Tool(name='sampling_tool', title=None, description='\n    大模型sampling服务端开发\n    :param ctx: 上下文类\n    :return: 处理完成\n    ', inputSchema={'properties': {}, 'title': 'sampling_toolArguments', 'type': 'object'}, outputSchema=None, icons=None, annotations=None, meta=None)]
context: RequestContext(request_id=0, meta=None, session=<mcp.client.session.ClientSession object at 0x000001B7CA8F76E0>, lifespan_context=None, request=None)
params: meta=None messages=[SamplingMessage(role='user', content=TextContent(type='text', text='帮我编写一个小学放假通知的文章，字数在100-200字之间。', annotations=None, meta=None))] modelPreferences=None systemPrompt=None includeContext=None temperature=None maxTokens=1024 stopSequences=None metadata=None
==== deepseek 决策结果： ChatCompletion(id='chatcmpl-234e81f1-bd1f-457c-ad9c-6cc4515b5076', choices=[Choice(finish_reason='stop', index=0, logprobs=None, message=ChatCompletionMessage(content='各位家长朋友：  \n根据学校安排，本学期寒假将于2025年1月20日开始，2月16日正式开学。请家长们合理安排孩子的假期生活，注意安全，督促孩子完成假期作业，保持良好的学习和生活习惯。感谢大家的支持与配合！  \n\n**XX小学**  \n2025年1月5日', refusal=None, role='assistant', annotations=None, audio=None, function_call=None, tool_calls=None))], created=1762422480, model='qwen-turbo', object='chat.completion', service_tier=None, system_fingerprint=None, usage=CompletionUsage(completion_tokens=76, prompt_tokens=33, total_tokens=109, completion_tokens_details=None, prompt_tokens_details=PromptTokensDetails(audio_tokens=None, cached_tokens=0)))
-----------调用工具返回结果： meta=None content=[TextContent(type='text', text='大模型sampling回调完成', annotations=None, meta=None)] structuredContent=None isError=False

1. 客户端

连接服务器时，创建ClientSession注册采样回调函数，ampling_callback=self.sampling_handler

sampling_handler函数，接受服务端sampling消息，转给大模型，让大模型返回结果，然后返回结果给服务端。

2. 服务端

@mcp.tool() 装饰工具函数，函数被调用时，函数会发送采样消息，含用户请求内容，等待采样结果。

流程：

客户端连接服务器，获取所有可用工具，然后调用mcp服务工具。

服务端接受到请求，在内部通过ctx.session.create_message 发送一条采样消息：

帮我编写一个小学放假通知的文章，字数在100-200字之间

客户端回调函数接收到请求，将请求转发给大模型，由大模型回答得到结果:

各位家长朋友：  \n根据学校安排，本学期寒假将于2025年1月20日开始，......

客户端CreateMessageResult将结果返回给服务端tool函数。