演示功能介绍

把用户内容按照提示词要求，转为一份通知单。

生成通知消息：
社区夏日纳凉晚会通知案例：梧桐树下的仲夏夜之梦
傍晚六点，李阿姨的帆布包还没放下，就被社区公告栏前的人群围住了。红底黄字的海报边角被风吹得卷起，上面的卡通月亮正朝她眨眼睛 ——“梧桐社区 2024 夏日纳凉晚会” 的标题下，彩色粉笔勾勒的吉他和冰淇淋甜筒歪歪扭扭。
“闺女，快帮我念念。” 穿碎花衬衫的张奶奶把老花镜推到头顶，手指戳着海报角落。我凑近看见打印的通知正文：“7 月 20 日晚 7 点，在社区中心小广场举办纳凉晚会，有歌舞表演、亲子游戏和露天电影。各家可自带零食，现场提供酸梅汤和西瓜。报名找 3 栋王阿姨，截止到 15 号……” 突然有穿蓝色校服的男孩挤进来，手里举着手机录像：“我要发给我妈看，她说要带自制的绿豆糕！”
回家路上，路灯把梧桐树影拉得老长。路过 4 栋时，听见王阿姨在楼道里喊：“二楼的陈师傅，您家孙女的古筝节目报上名啦！” 防盗门 “吱呀” 开了条缝，白发爷爷探出头：“几点开始啊？我得提前搬折叠椅占位置。”“跟通知上一样，七点开始，电影放到十点呢！” 王阿姨的声音混着锅铲翻炒声飘出来，“对了，您记得带驱蚊水！”
晚会当天傍晚，小广场的彩灯刚亮起，就有家长扛着野餐垫来了。穿汉服的小姑娘蹲在舞台边，往脸上贴星星贴纸，她妈妈举着通知复印件核对：“节目单上第三个就是你的琵琶独奏，别紧张啊。” 卖棉花糖的大叔推车经过，糖丝在暮色里拉出银线，有小孩追着喊：“叔叔，我要草莓味的！”
当社区合唱团的《茉莉花》响起时，我看见张奶奶正跟着节奏晃扇子。月光穿过梧桐叶隙落在她肩上，旁边的小男孩突然举起荧光棒：“奶奶你看，像不像通知里说的‘星光晚会’？” 远处的电影幕布亮起，《龙猫》的旋律混着此起彼伏的笑声，飘向挂满灯笼的树梢 —— 原来通知里那句 “带上好心情来赴约”，真的有人悄悄记在了心上。

最终的结果：

 活动主题：社区夏日纳凉晚会  
参与对象：社区居民及家庭  
时间安排：7月20日晚7点至十点  
举办地点：社区中心小广场  
报名方式：联系3栋王阿姨，截止到15号

服务端

编写提示词模版

from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Prompt MCP Server")

@mcp.prompt()
def notify_prompt(notify:str):
    '''
    这个方法是一个提示方法，总结概要信息
    : notify: 文章内容
    : return: 提示信息
    '''
    return {
        "role":"user",
          "content": f"""
                    这是活动通知内容：“{notify}”，请按标准格式整理通知要点，规则如下：
                    提取活动关键信息。
                    每个信息点按以下格式总结：
                    活动主题：需体现活动核心内容与目的
                    参与对象：明确可参加的人群或范围
                    时间安排：活动开始至结束的具体时段
                    举办地点：活动实际开展的地址
                    报名方式：参与活动的申请途径及截止时间
                    总结语言需简洁明了，避免使用复杂句式。
         """
    }
if __name__ == "__main__":
    print(" Prompt MCP Server is running...")
    mcp.run(transport="sse")

客户端

"""4.2 【资源类】Resource 客户端开发"""
import asyncio
import json
from contextlib import AsyncExitStack

from mcp import StdioServerParameters, stdio_client, ClientSession, ListResourcesResult, ListPromptsResult
from mcp.client.sse import sse_client
from openai import OpenAI


class MCPClient:

    def __init__(self):
        self.async_exit_stack = AsyncExitStack()
        self.session = None
        self.deepseek = OpenAI(
            api_key="sk-3ad75cfac6b***159ab00f8d3d473b2a",
            base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
        )
        self.prompts = None

    async def connect_to_server(self, url: str):
        # 一 、创建stdio_client
        client = sse_client(url=url)
        transport = await self.async_exit_stack.enter_async_context(client)
        read_stream, write_stream = transport
        # 二、创建会话client
        client_session = ClientSession(read_stream, write_stream)
        self.session = await self.async_exit_stack.enter_async_context(client_session)
        # 三、初始化会话
        await self.session.initialize()

    async def execute(self, question: str):
        # 一、获取server服务端中的提示类list_prompt列表 (需要改动）
        list_prompt: ListPromptsResult = await self.session.list_prompts()
        prompts = list_prompt.prompts
        print("打印出获取的工具列表：", prompts)
        self.prompts = {}
        # 二、创建function calling 格式(大模型使用）、
        prompts_tools = []
        for resource in prompts:
            name = resource.name
            description = resource.description
            prompts_tools.append(
                {
                    "type": "function",
                    "function": {
                        "name": name,
                        "description": description,
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "notify": {
                                    "type": "string",
                                    "description": "通知内容"
                                }
                            },
                            "required": ["notify"]  # 告诉大模型这个字段必填
                        }
                    }
                }
            )
            # 初始化resources属性
            self.prompts[name] = {
                "name": name,
                "description": description,
                "mime_type": None
            }

        # 三、 创建messages,deepseek大模型的格式
        messages = [
            {
                "role": "user",
                "content": question
            }
        ]
        # 四、调用deepseek大模型
        deepseek_response = self.deepseek.chat.completions.create(
            model="qwen-turbo",
            messages=messages,
            tools=prompts_tools
        )
        # 打印出大模型的决策结果
        print("==== deepseek 响应持结果：", deepseek_response)
        choice_result = deepseek_response.choices[0]
        # 第二次调用大模型的前置参数
        messages.append(choice_result.message.model_dump())
        tool_call = choice_result.message.tool_calls[0]

        print(" tool_call:", tool_call)
        print("大模型决策的最终结果,工具名称：", tool_call.function.name, ",参数：", tool_call.function.arguments)
        function_name = tool_call.function.name
        arguments = json.loads(tool_call.function.arguments)
        # arguments = {
        #     "notify": question  # 键=参数名，值=字符串
        # }
        # 五、调用工具链 （需要改动）
        prompts_result = await self.session.get_prompt(
            name=function_name,
            arguments=arguments
        )
        print("==== 工具调用结果：", prompts_result)
        # 最终的结果
        result = prompts_result.messages[0].content.text
        print("==== 最终的结果：", result)

        # 六、使用大模型生成最终的结果，并且使用语言模型生成最终的结果
        messages.append({
            "role": "tool",
            "content": result,
            "tool_call_id": tool_call.id
        })
        # 再次调用大模型
        deepseek_response = self.deepseek.chat.completions.create(
            model="qwen-turbo",
            messages=messages,
            tools=prompts_tools,
        )
        # 获取最终的结果
        result = deepseek_response.choices[0].message.content
        print("==== 最终的结果：", result)

    # 关闭资源
    async def aclose(self):
        await self.async_exit_stack.aclose()


async def main():
    client = MCPClient()
    try:
        await client.connect_to_server("http://localhost:8000/sse")
        with open("prompt.data", "r", encoding="utf-8") as f:
            content = f.read()
            await client.execute(content)
    except Exception as e:
        print(f"连接失败: {e}")
        return
    finally:
        await client.aclose()


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

1 第一次向大模型对话

messages = [
            {
                "role": "user",
                "content": question
            }
        ]

deepseek_response = self.deepseek.chat.completions.create(
            model="qwen-turbo",
            messages=messages,
            tools=prompts_tools
        )

prompts_tools.append(
                {
                    "type": "function",
                    "function": {
                        "name": name,
                        "description": description,
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "notify": {
                                    "type": "string",
                                    "description": "通知内容"
                                }
                            },
                            "required": ["notify"]  # 告诉大模型这个字段必填
                        }
                    }
                }

参数 messages : 传递一条message：角色是user，内容是  ‘文章内容’。

参数 tools：传递一个工具列表，模型可以调用这些工具。可以提供自定义工具或者功能工具。

        其中的parameters是 以 JSON Schema 对象的形式描述。

大模型返回结果：

大模型决定需要调用tool，需要调用的函数名称，以及参数也帮你生成好了。

后续只需把 tool_calls[0].id 和这段结果通过 role="tool" 回传，即可让模型继续生成最终回复。

ChatCompletion(
  id='chatcmpl-8cc61083-434c-40f4-8eb0-bb1d65a02f32',
  model='qwen-turbo',
  created=1763025938,
  object='chat.completion',
  service_tier=None,
  system_fingerprint=None,

  choices=[
    Choice(
      index=0,
      finish_reason='tool_calls',
      logprobs=None,
      message=ChatCompletionMessage(
        role='assistant',
        content='',
        function_call=None,
        tool_calls=[
          ChatCompletionMessageFunctionToolCall(
            id='call_19050e78bae243ef87303c',
            type='function',
            function=Function(
              name='notify_prompt',
              arguments='{"notify":"社区夏日纳凉晚会通知案例：梧桐树下的仲夏夜之梦……（略）"}'
            )
          )
        ]
      )
    )
  ],

  usage=CompletionUsage(
    prompt_tokens=767,
    completion_tokens=302,
    total_tokens=1069,
    prompt_tokens_details=PromptTokensDetails(audio_tokens=None, cached_tokens=0),
    completion_tokens_details=None
  )
)

调用MCP提示词模版，生成完整提示词

上一步大模型决策返回了需要调用的函数名称和参数。

function_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
prompts_result = await self.session.get_prompt(
            name=function_name,
            arguments=arguments
        )

第二次对话大模型

        messages.append({
            "role": "tool",
            "content": result,
            "tool_call_id": tool_call.id
        })
        # 再次调用大模型
        deepseek_response = self.deepseek.chat.completions.create(
            model="qwen-turbo",
            messages=messages,
            tools=prompts_tools,
        )

追加messages：

字段	必须值	作用
role	"tool"	声明这是“工具返回值”，不是用户也不是助手。
content	result（字符串）	真正的执行结果，模型会读它再组织语言。
tool_call_id	tool_call.id	把结果跟之前那条工具调用一一对应；若写错会报 400。

大模型返回结果

ChatCompletion(
  id='chatcmpl-96785c8f-f969-4e07-bbe5-d524eb3854ed',
  model='qwen-turbo',
  created=1763027679,
  object='chat.completion',
  service_tier=None,
  system_fingerprint=None,

  choices=[
    Choice(
      index=0,
      finish_reason='stop',
      logprobs=None,
      message=ChatCompletionMessage(
        role='assistant',
        content='活动主题：社区夏日纳凉晚会  \n'
                '参与对象：社区居民及家庭  \n'
                '时间安排：7月20日晚7点至十点  \n'
                '举办地点：社区中心小广场  \n'
                '报名方式：联系3栋王阿姨，截止到15号',
        function_call=None,
        tool_calls=None
      )
    )
  ],

  usage=CompletionUsage(
    prompt_tokens=1096,
    completion_tokens=53,
    total_tokens=1149,
    prompt_tokens_details=PromptTokensDetails(audio_tokens=None, cached_tokens=0),
    completion_tokens_details=None
  )
)

The tool 调用流程：

1. 向模型发起请求，请求中包含模型可以调用的工具。

2. 接收模型发起的工具调用。

3. 在应用程序端根据工具调用的输入执行代码。

4. 使用工具的输出向模型发起第二次请求。

5. 接收模型的最终响应（或更多工具调用）。