三亩地 三亩地SAN MU DI · CODE DIARY
交流投稿
ARTICLE DETAIL

日记详情

真实记录编程学习的某一天，欢迎挑你感兴趣的翻一翻。

首页/学习日记/日记详情

MCP和LLM对话验证

发布时间：2026/8/11 9:44:44 三亩地 编程学习日记
MCP和LLM对话验证

MCP和LLM对话验证

MCP和LLM对话验证，验证在LLM执行MCP过程中工具存在状态，验证LLM是如何完成工具的调用的！

0-MCP学习什么

  • 1-【理论】先知道MCP是什么东西（FunctionCall标准化对接，一个协议，只要大家都按照这个标注去写，就可以把这个标准推广到所有的LLM）

  • 2-【理论】MCP 只是一个理论。如果说仅有一个理论，没有一个框架把它落地，那MCP就是空谈->于是有人开发了【FastMcp】，使用标准就可以把Service暴露出去

    • @mcp.tool()
    • @mcp.prompt()
    • @mcp.resource(“config://app/settings”)

  • 3-【实战】学习使用fastMCP编写MCP，参考：https://gitee.com/enzoism/python_mcp_inspector

  • 4-【实战】【langchain框架】-观察MCP在LLM对话过程中工作原理，【转化为FunctionCall的Prompt描述给到LLM】+【死循环】，参考： https://gitee.com/enzoism/mcp_langchain_demo

  • 5-【实战】【openai框架】-开发一个前端，支持配置外部MCP，后端来处理LLM的Tool和对话，参考：https://gitee.com/enzoism/python_mcp_02_openai_agent

  • 6-【拓展】如果你想快速验证MCP，没有必要使用我上面的代码进行MCP验证，你可以使用【Inspector】或者【CherryStudio】脚本【npx @modelcontextprotocol/inspector python src/server.py】进行调试

  • 7-【理论】MCP知识整理参考：https://gitee.com/enzoism/knowledge_2026_mcp

1-验证代码

  • 1）步骤一：运行MC项目暴露API接口：https://gitee.com/enzoism/python_mcp_inspector
  • 2）步骤二：运行MCP项目测试，配置MCP后进行验证-https://gitee.com/enzoism/mcp_langchain_demo

2-验证结果

1-FastMcp代码编写示例

""" MCP Demo Server (Python) 基于 mcp Python SDK 实现，支持 Tools、Resources、Prompts 三大核心能力。 """ from mcp.server.fastmcp import FastMCP # 初始化 MCP 服务实例 mcp = FastMCP("demo-server") # ============================================ # 1. 注册工具能力：天气查询、数学计算、时间获取 # ============================================ @mcp.tool() async def get_weather(city: str) -> str: """获取指定城市的当前天气信息 Args: city: 城市名称，如 '北京' """ mock_data = { "北京": {"temp": "22°C", "weather": "晴", "humidity": "45%"}, "上海": {"temp": "25°C", "weather": "多云", "humidity": "60%"}, "深圳": {"temp": "28°C", "weather": "阵雨", "humidity": "80%"}, } data = mock_data.get(city) if data: return f"城市: {city}\n温度: {data['temp']}\n天气: {data['weather']}\n湿度: {data['humidity']}" return f"未找到城市 '{city}' 的天气数据" @mcp.tool() async def calculate(expression: str) -> str: """安全地计算数学表达式 Args: expression: 数学表达式，如 '2 + 3 * 4' """ allowed_chars = set("0123456789+-*/().% ") if not all(c in allowed_chars for c in expression): return "错误：表达式包含不允许的字符" try: # 使用受限的 eval 环境，只允许数学运算 result = eval(expression, {"__builtins__": {}}, {}) return str(result) except Exception as e: return f"计算错误：{e}" @mcp.tool() async def get_current_time() -> str: """获取当前系统时间""" from datetime import datetime now = datetime.now() return now.strftime("%Y-%m-%d %H:%M:%S") # ============================================ # 2. 注册资源能力：应用配置读取、本地文件读取 # ============================================ @mcp.resource("config://app/settings") async def app_settings() -> str: """获取应用配置信息""" import json config = { "app_name": "Demo MCP Server", "version": "1.0.0", "features": {"weather": True, "calculator": True, "file_reader": True}, "max_file_size_mb": 10, } return json.dumps(config, indent=2, ensure_ascii=False) @mcp.resource("file://docs/readme") async def readme() -> str: """项目说明文档""" return "这是一个基于 MCP 协议的演示服务端（Python版），支持天气查询、数学计算、代码审查等功能。" # ============================================ # 3. 注册提示词模板：代码审查、错误分析 # ============================================ @mcp.prompt() def code_review(code: str, language: str = "python") -> str: """代码审查提示词 Args: code: 待审查的代码 language: 编程语言，默认为 python """ return ( f"请对以下 {language} 代码进行审查，" f"从代码质量、潜在Bug、性能、安全性、最佳实践五个维度给出评价和优化建议：\n" f"```{language}\n{code}\n```" ) @mcp.prompt() def error_analysis(error: str, context: str = "") -> str: """错误分析提示词 Args: error: 错误信息或堆栈跟踪 context: 相关上下文代码 """ prompt = f"请分析以下错误信息，给出可能的原因和解决方案：\n\n错误信息：\n```\n{error}\n```" if context: prompt += f"\n\n相关代码：\n```\n{context}\n```" return prompt # ============================================ # 启动服务，采用 stdio 本地传输模式 # ============================================ if __name__ == "__main__": # 先设置配置 mcp.settings.host = "0.0.0.0" mcp.settings.port = 8080 # 运行sse模式 mcp.run(transport="sse")

2-MCP将@McpTool标记的API转化为FunctionCall提供给LLM进行选择

  • 配置外部的MCP
asyncdefsetup_mcp_tools(self,mcp_configs:List[Dict[str,Any]]):"""设置MCP工具"""frommodelsimportMCPServiceConfig"""尝试mcp管理器会和你配置的MCP列表中每一个MCP进行连接"""configs=[MCPServiceConfig(**config)forconfiginmcp_configs]awaitself.mcp_manager.connect_services(configs)"""从MCP列表中获取每一个MCP的Tools添加到列表中"""tools=awaitself.mcp_manager.get_all_tools()logger.info(f"Loaded{len(tools)}MCP tools")returntools

  • 和LLM对话的死循环

  • 1-先连接MCP服务获取工具信息mcp_tools_list

    2-绑定工具到LLM（就是把这些tool拼接为prompt，告诉LLM可以从这些tool列表中选择需要的Tool）

    3-然后就是构建一个死循环

    • 如果命中了tool就获取tool的结果（while hasattr(response, ‘tool_calls’) and response.tool_calls and iteration < max_iterations）
    • 否则就尝试获取LLM的问答结果mcp_results
asyncdefchat_with_mcp_context(self,messages:List[ChatMessage],mcp_configs:List[Dict[str,Any]],model_name:str="qwen-plus")->Dict[str,Any]:"""使用MCP上下文进行对话"""ifnotself.llm:self.init_llm(model_name)# 先连接MCP服务获取工具信息mcp_tools_list=awaitself.setup_mcp_tools(mcp_configs)# 创建LangChain工具langchain_tools=[]formcp_toolinmcp_tools_list:tool_name=mcp_tool["function"]["name"]tool_desc=mcp_tool["function"]["description"]tool_params=mcp_tool["function"].get("parameters",{})langchain_tool=self.create_langchain_tool(tool_name,tool_desc,tool_params)langchain_tools.append(langchain_tool)self.tools_map[tool_name]=langchain_tool logger.info(f"Created{len(langchain_tools)}LangChain tools")# 绑定工具到LLMllm_with_tools=self.llm.bind_tools(langchain_tools)# 格式化消息formatted_messages=self._format_messages(messages)system_prompt="""你是一个AI助手，可以使用MCP（Model Context Protocol）服务提供的工具来获取信息或执行操作。 当用户的问题需要使用工具时，请调用相应的工具。工具执行结果会自动返回给你。 请根据工具返回的结果给出准确的回答。如果工具执行失败，请向用户说明情况。"""full_messages=[SystemMessage(content=system_prompt)]+formatted_messages# 第一次调用LLMresponse=awaitllm_with_tools.ainvoke(full_messages)mcp_results=[]iteration=0max_iterations=5# 防止无限循环# 检查是否有工具调用whilehasattr(response,'tool_calls')andresponse.tool_callsanditeration<max_iterations:iteration+=1logger.info(f"Iteration{iteration}: Found{len(response.tool_calls)}tool calls")# 添加AI响应到消息历史full_messages.append(response)# 执行所有工具调用fortool_callinresponse.tool_calls:tool_name=tool_call["name"]tool_args=tool_call["args"]tool_id=tool_call.get("id","")logger.info(f"Executing tool:{tool_name}with args:{tool_args}")try:# 从工具映射中获取工具并执行iftool_nameinself.tools_map:tool_obj=self.tools_map[tool_name]result=awaittool_obj.ainvoke(tool_args)logger.info(f"Tool result:{str(result)[:200]}")mcp_results.append({"tool":tool_name,"arguments":tool_args,"result":str(result),"hit":True})# 添加工具结果消息full_messages.append(ToolMessage(content=str(result),tool_call_id=tool_id))else:error_msg=f"Tool{tool_name}not found"mcp_results.append({"tool":tool_name,"arguments":tool_args,"error":error_msg,"hit":False})full_messages.append(ToolMessage(content=error_msg,tool_call_id=tool_id))exceptExceptionase:error_msg=f"Tool execution error:{str(e)}"logger.error(error_msg)mcp_results.append({"tool":tool_name,"arguments":tool_args,"error":error_msg,"hit":False})full_messages.append(ToolMessage(content=error_msg,tool_call_id=tool_id))# 再次调用LLM处理工具结果response=awaitllm_with_tools.ainvoke(full_messages)# 如果有工具调用，返回最终结果ifmcp_results:return{"content":response.content,"mcp_calls":mcp_results,"mcp_context_used":True}# 没有工具调用，直接返回响应return{"content":response.content,"mcp_calls":[],"mcp_context_used":False}

3-简单验证

上面的验证，使用了2个项目，简单点直接使用后端接口进行验证，验证MCP资源的使用-后端代码【openai框架】-https://gitee.com/enzoism/python_mcp_02_openai_agent

importasyncioimportosimportshutilimportsubprocessimporttimefromtypingimportAnyfromagentsimportAgent,Runner,RunConfig,set_tracing_disabledfromagents.mcpimportMCPServer,MCPServerSsefromagents.model_settingsimportModelSettingsfromdotenvimportload_dotenvfrommodel_providers.deepseekimportDeepSeekModelProvider# 1-环境变量加载相关load_dotenv()BASE_URL=os.getenv("BASE_URL")or""API_KEY=os.getenv("API_KEY")or""MODEL_NAME=os.getenv("MODEL_NAME")or""ifnotBASE_URLornotAPI_KEYornotMODEL_NAME:raiseValueError("请通过环境变量或代码设置EXAMPLE_BASE_URL、EXAMPLE_API_KEY、EXAMPLE_MODEL_NAME。")# 2-跳过大模型的链路追踪set_tracing_disabled(disabled=True)""" 本例使用自定义提供程序调用Runner.run()的部分，并直接调用OpenAI进行其他操作。 步骤： 1. 【实例化LLM】ModelProvider对象-并构建RunConfig 2. 【实例化Agent】创建一个Agent。 3. 在调用Runner.run()结合【LLM】+【Agent】进行问答 - 1）直接和大模型对话 - 2）调用MCPServer_SSE模式-[MCP工具1-数字加和] - 3）调用MCPServer_SSE模式-[MCP工具2-随机选定的字符] """asyncdefrun_mcp_server(mcp_server:MCPServer):# 1-【实例化LLM】ModelProvider对象-并构建RunConfigrun_config=RunConfig(model_provider=DeepSeekModelProvider(BASE_URL,API_KEY,MODEL_NAME))# 2-【实例化Agent】创建一个Agentagent=Agent(name="Assistant",instructions="使用工具回答大模型的问题",mcp_servers=[mcp_server],model_settings=ModelSettings(tool_choice="required"),)# 3.1-直接和大模型对话message="给我讲一个笑话吧！"print(f"\n\n【大模型请求案例】->{message}")result=awaitRunner.run(starting_agent=agent,input=message,run_config=run_config)print(result.final_output)# 3.2-调用MCPServer_SSE模式-[MCP工具1-数字加和]message="What's the weather in Tokyo?"print(f"\n\n【大模型请求案例】->{message}")result=awaitRunner.run(starting_agent=agent,input=message,run_config=run_config)print(result.final_output)# 3.3-调用MCPServer_SSE模式-[MCP工具2-随机选定的字符]message="What's the secret word?"print(f"\n\n【大模型请求案例】->{message}")result=awaitRunner.run(starting_agent=agent,input=message,run_config=run_config)print(result.final_output)asyncdefinit_mcp_server()->MCPServerSse:# 1-创建 MCP 服务器连接实例，但不立即运行weather_server=MCPServerSse(name="SSE Python Server",params={"url":"http://localhost:8000/sse",},# 缓存工具列表以减少延迟，不需要每次连接时重新获取工具列表cache_tools_list=True)# 2-手动连接到MCP服务器print("正在连接到MCP服务器...")awaitweather_server.connect()print("MCP服务器连接成功！")# 3-等待服务器连接成功并获取MCP服务可用工具列表tools=awaitweather_server.list_tools()print("\n可用工具列表: ")fortoolintools:print(f" -{tool.name}:{tool.description}")returnweather_serverif__name__=="__main__":# 1-让我们确保用户已安装 uvifnotshutil.which("uv"):raiseRuntimeError("uv is not installed. Please install it: https://docs.astral.sh/uv/getting-started/installation/")# 2-我们将以子进程运行SSE服务器。通常这会是远程服务器，但在这个演示中，我们将在本地运行，地址为http://localhost:8000/sseprocess:subprocess.Popen[Any]|None=Nonetry:# 2.1-运行 `uv run server.py` 以启动 SSE 服务器print("Starting SSE server at http://localhost:8000/sse ...")this_dir=os.path.dirname(os.path.abspath(__file__))mcp_server_file=os.path.join(this_dir,"mcp02_sse.py")process=subprocess.Popen(["uv","run",mcp_server_file])# 2.2-给它3秒钟启动time.sleep(3)print("SSE server started. Running example...\n\n")exceptExceptionase:print(f"Error starting SSE server:{e}")exit(1)# 3-运行示例try:# 1-获取McpServerserver=asyncio.run(init_mcp_server())# 2-运行McpServerasyncio.run(run_mcp_server(server))finally:ifprocess:process.terminate()
编程学习 学习日记 实战经验
← 返回列表