MCP之Prompt提示词工程

演示功能介绍

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

生成通知消息：
社区夏日纳凉晚会通知案例：梧桐树下的仲夏夜之梦
傍晚六点，李阿姨的帆布包还没放下，就被社区公告栏前的人群围住了。红底黄字的海报边角被风吹得卷起，上面的卡通月亮正朝她眨眼睛 ——“梧桐社区 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 AsyncExitStackfrom mcp import StdioServerParameters, stdio_client, ClientSession, ListResourcesResult, ListPromptsResult
from mcp.client.sse import sse_client
from openai import OpenAIclass MCPClient:def __init__(self):self.async_exit_stack = AsyncExitStack()self.session = Noneself.deepseek = OpenAI(api_key="sk-3ad75cfac6b***159ab00f8d3d473b2a",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")self.prompts = Noneasync def connect_to_server(self, url: str):# 一 、创建stdio_clientclient = sse_client(url=url)transport = await self.async_exit_stack.enter_async_context(client)read_stream, write_stream = transport# 二、创建会话clientclient_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.promptsprint("打印出获取的工具列表：", prompts)self.prompts = {}# 二、创建function calling 格式(大模型使用）、prompts_tools = []for resource in prompts:name = resource.namedescription = resource.descriptionprompts_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.namearguments = 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.textprint("==== 最终的结果：", 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.contentprint("==== 最终的结果：", 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}")returnfinally: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. 接收模型的最终响应（或更多工具调用）。