MCP和LLM对话验证

MCP和LLM对话验证MCP和LLM对话验证验证在LLM执行MCP过程中工具存在状态验证LLM是如何完成工具的调用的0-MCP学习什么1-【理论】先知道MCP是什么东西FunctionCall标准化对接一个协议只要大家都按照这个标注去写就可以把这个标准推广到所有的LLM2-【理论】MCP 只是一个理论。如果说仅有一个理论没有一个框架把它落地那MCP就是空谈-于是有人开发了【FastMcp】使用标准就可以把Service暴露出去mcp.tool()mcp.prompt()mcp.resource(“config://app/settings”)3-【实战】学习使用fastMCP编写MCP参考https://gitee.com/enzoism/python_mcp_inspector4-【实战】【langchain框架】-观察MCP在LLM对话过程中工作原理【转化为FunctionCall的Prompt描述给到LLM】【死循环】参考 https://gitee.com/enzoism/mcp_langchain_demo5-【实战】【openai框架】-开发一个前端支持配置外部MCP后端来处理LLM的Tool和对话参考https://gitee.com/enzoism/python_mcp_02_openai_agent6-【拓展】如果你想快速验证MCP没有必要使用我上面的代码进行MCP验证你可以使用【Inspector】或者【CherryStudio】脚本【npx modelcontextprotocol/inspector python src/server.py】进行调试7-【理论】MCP知识整理参考https://gitee.com/enzoism/knowledge_2026_mcp1-验证代码1步骤一运行MC项目暴露API接口https://gitee.com/enzoism/python_mcp_inspector2步骤二运行MCP项目测试配置MCP后进行验证-https://gitee.com/enzoism/mcp_langchain_demo2-验证结果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, indent2, ensure_asciiFalse) 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(transportsse)2-MCP将McpTool标记的API转化为FunctionCall提供给LLM进行选择配置外部的MCPasyncdefsetup_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添加到列表中toolsawaitself.mcp_manager.get_all_tools()logger.info(fLoaded{len(tools)}MCP tools)returntools和LLM对话的死循环1-先连接MCP服务获取工具信息mcp_tools_list2-绑定工具到LLM就是把这些tool拼接为prompt告诉LLM可以从这些tool列表中选择需要的Tool3-然后就是构建一个死循环如果命中了tool就获取tool的结果while hasattr(response, ‘tool_calls’) and response.tool_calls and iteration max_iterations否则就尝试获取LLM的问答结果mcp_resultsasyncdefchat_with_mcp_context(self,messages:List[ChatMessage],mcp_configs:List[Dict[str,Any]],model_name:strqwen-plus)-Dict[str,Any]:使用MCP上下文进行对话ifnotself.llm:self.init_llm(model_name)# 先连接MCP服务获取工具信息mcp_tools_listawaitself.setup_mcp_tools(mcp_configs)# 创建LangChain工具langchain_tools[]formcp_toolinmcp_tools_list:tool_namemcp_tool[function][name]tool_descmcp_tool[function][description]tool_paramsmcp_tool[function].get(parameters,{})langchain_toolself.create_langchain_tool(tool_name,tool_desc,tool_params)langchain_tools.append(langchain_tool)self.tools_map[tool_name]langchain_tool logger.info(fCreated{len(langchain_tools)}LangChain tools)# 绑定工具到LLMllm_with_toolsself.llm.bind_tools(langchain_tools)# 格式化消息formatted_messagesself._format_messages(messages)system_prompt你是一个AI助手可以使用MCPModel Context Protocol服务提供的工具来获取信息或执行操作。 当用户的问题需要使用工具时请调用相应的工具。工具执行结果会自动返回给你。 请根据工具返回的结果给出准确的回答。如果工具执行失败请向用户说明情况。full_messages[SystemMessage(contentsystem_prompt)]formatted_messages# 第一次调用LLMresponseawaitllm_with_tools.ainvoke(full_messages)mcp_results[]iteration0max_iterations5# 防止无限循环# 检查是否有工具调用whilehasattr(response,tool_calls)andresponse.tool_callsanditerationmax_iterations:iteration1logger.info(fIteration{iteration}: Found{len(response.tool_calls)}tool calls)# 添加AI响应到消息历史full_messages.append(response)# 执行所有工具调用fortool_callinresponse.tool_calls:tool_nametool_call[name]tool_argstool_call[args]tool_idtool_call.get(id,)logger.info(fExecuting tool:{tool_name}with args:{tool_args})try:# 从工具映射中获取工具并执行iftool_nameinself.tools_map:tool_objself.tools_map[tool_name]resultawaittool_obj.ainvoke(tool_args)logger.info(fTool result:{str(result)[:200]})mcp_results.append({tool:tool_name,arguments:tool_args,result:str(result),hit:True})# 添加工具结果消息full_messages.append(ToolMessage(contentstr(result),tool_call_idtool_id))else:error_msgfTool{tool_name}not foundmcp_results.append({tool:tool_name,arguments:tool_args,error:error_msg,hit:False})full_messages.append(ToolMessage(contenterror_msg,tool_call_idtool_id))exceptExceptionase:error_msgfTool 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(contenterror_msg,tool_call_idtool_id))# 再次调用LLM处理工具结果responseawaitllm_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_agentimportasyncioimportosimportshutilimportsubprocessimporttimefromtypingimportAnyfromagentsimportAgent,Runner,RunConfig,set_tracing_disabledfromagents.mcpimportMCPServer,MCPServerSsefromagents.model_settingsimportModelSettingsfromdotenvimportload_dotenvfrommodel_providers.deepseekimportDeepSeekModelProvider# 1-环境变量加载相关load_dotenv()BASE_URLos.getenv(BASE_URL)orAPI_KEYos.getenv(API_KEY)orMODEL_NAMEos.getenv(MODEL_NAME)orifnotBASE_URLornotAPI_KEYornotMODEL_NAME:raiseValueError(请通过环境变量或代码设置EXAMPLE_BASE_URL、EXAMPLE_API_KEY、EXAMPLE_MODEL_NAME。)# 2-跳过大模型的链路追踪set_tracing_disabled(disabledTrue) 本例使用自定义提供程序调用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_configRunConfig(model_providerDeepSeekModelProvider(BASE_URL,API_KEY,MODEL_NAME))# 2-【实例化Agent】创建一个AgentagentAgent(nameAssistant,instructions使用工具回答大模型的问题,mcp_servers[mcp_server],model_settingsModelSettings(tool_choicerequired),)# 3.1-直接和大模型对话message给我讲一个笑话吧print(f\n\n【大模型请求案例】-{message})resultawaitRunner.run(starting_agentagent,inputmessage,run_configrun_config)print(result.final_output)# 3.2-调用MCPServer_SSE模式-[MCP工具1-数字加和]messageWhats the weather in Tokyo?print(f\n\n【大模型请求案例】-{message})resultawaitRunner.run(starting_agentagent,inputmessage,run_configrun_config)print(result.final_output)# 3.3-调用MCPServer_SSE模式-[MCP工具2-随机选定的字符]messageWhats the secret word?print(f\n\n【大模型请求案例】-{message})resultawaitRunner.run(starting_agentagent,inputmessage,run_configrun_config)print(result.final_output)asyncdefinit_mcp_server()-MCPServerSse:# 1-创建 MCP 服务器连接实例但不立即运行weather_serverMCPServerSse(nameSSE Python Server,params{url:http://localhost:8000/sse,},# 缓存工具列表以减少延迟不需要每次连接时重新获取工具列表cache_tools_listTrue)# 2-手动连接到MCP服务器print(正在连接到MCP服务器...)awaitweather_server.connect()print(MCP服务器连接成功)# 3-等待服务器连接成功并获取MCP服务可用工具列表toolsawaitweather_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]|NoneNonetry:# 2.1-运行 uv run server.py 以启动 SSE 服务器print(Starting SSE server at http://localhost:8000/sse ...)this_diros.path.dirname(os.path.abspath(__file__))mcp_server_fileos.path.join(this_dir,mcp02_sse.py)processsubprocess.Popen([uv,run,mcp_server_file])# 2.2-给它3秒钟启动time.sleep(3)print(SSE server started. Running example...\n\n)exceptExceptionase:print(fError starting SSE server:{e})exit(1)# 3-运行示例try:# 1-获取McpServerserverasyncio.run(init_mcp_server())# 2-运行McpServerasyncio.run(run_mcp_server(server))finally:ifprocess:process.terminate()