ai调用mcp 直接基于mcp库

不过ai 直接调用mcp方法 ClientSession, StdioServerParameters-CSDN博客

https://www.bilibili.com/video/BV1ZLkLB1EWo

import tkinter as tk from tkinter import scrolledtext, messagebox import asyncio import sys import os import json import threading import re sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) class MCPAICaller: def __init__(self, root): self.root = root # 去除窗口标题栏和边框 self.root.overrideredirect(True) # 设置窗口位置和大小 - 改为更小的窗口 self.root.geometry("600x400+100+100") # 设置更高的透明度 (从0.95改为0.98) self.root.attributes('-alpha', 0.98) # 动态导入LLM和VLM服务 import importlib.util llm_spec = importlib.util.spec_from_file_location( "llm_class", os.path.join(os.path.dirname(os.path.dirname(__file__)), "llm_server", "llm_class.py") ) llm_module = importlib.util.module_from_spec(llm_spec) llm_spec.loader.exec_module(llm_module) self.LLMService = llm_module.LLMService self.VLMService = llm_module.VLMService # 实例化LLM和VLM服务 self.llm_service = self.LLMService() self.vlm_service = self.VLMService() # 实例化MCP客户端 from mcp_cline.mcp_client import MCPClient self.mcp_client = MCPClient() # 延迟加载MCP工具列表 - 初始化为None表示尚未加载 self.available_tools = None self.method_map = self.generate_method_map() self.setup_ui() def load_mcp_tools_on_demand(self): """ 首次需要工具列表时才加载(懒加载) """ if self.available_tools is not None: return # 已经加载过了,无需重复加载 try: all_tools = {} # 遍历MCPClient中所有的服务器 for server_name in self.mcp_client.servers.keys(): try: # 为每个服务器异步获取工具列表 tools = asyncio.run(self.mcp_client.list_tools(server_name)) # 将Tool对象转换为字典格式以便于使用 if tools: tool_dict = {} for tool in tools: tool_dict[tool.name] = { 'name': tool.name, 'description': tool.description } all_tools[server_name] = tool_dict else: all_tools[server_name] = {} except Exception as e: print(f"获取服务器 {server_name} 的工具列表失败: {str(e)}") continue self.available_tools = all_tools print(f"已按需加载MCP工具列表: {list(all_tools.keys())}") except Exception as e: print(f"按需加载MCP工具列表失败: {str(e)}") self.available_tools = {} def generate_method_map(self): """ 生成方法编号映射 """ method_map = {} idx = 1 # 查看帮助 method_map[idx] = {"name": "查看帮助", "description": "输入 /h 可查看所有MCP工具"} idx += 1 # MCP工具调用 method_map[idx] = {"name": "MCP工具调用", "description": "AI可根据上下文自动调用MCP工具"} idx += 1 # 添加MCP工具 - 仅当工具已加载时才添加 if self.available_tools is not None and self.available_tools: for server_name, tools in self.available_tools.items(): if tools: for tool_name, tool_info in tools.items(): method_map[idx] = { "name": f"调用{tool_name}", "description": f"{tool_info['description']} (服务器: {server_name})" } idx += 1 return method_map def setup_ui(self): # 主框架 - 半透明背景 main_frame = tk.Frame(self.root, bg='#f0f0f0', bd=0, highlightthickness=0) main_frame.pack(fill=tk.BOTH, expand=True) # 对话框容器 dialog_container = tk.Frame(main_frame, bg='#f0f0f0', bd=0, highlightthickness=0) dialog_container.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) dialog_container.columnconfigure(0, weight=1) dialog_container.rowconfigure(0, weight=1) # 输出区域 dialog_container.rowconfigure(1, weight=0) # 输入区域 dialog_container.rowconfigure(2, weight=0) # 按钮区域 # 输出区域 - 半透明背景,黑色文字 self.output_text = scrolledtext.ScrolledText( dialog_container, height=12, # 减少高度 bg='#ffffff', # 白色背景 fg='#333333', # 深灰色文字 bd=0, # 无边框 highlightthickness=0, # 无高亮边框 wrap=tk.WORD, font=('Arial', 10), # 减小字体 state=tk.NORMAL ) self.output_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S), pady=(0, 10)) # 输入区域 - 明确的输入框,白色背景,黑色文字 self.input_text = scrolledtext.ScrolledText( dialog_container, height=3, # 减少高度 bg='#ffffff', # 白色背景,清晰可见 fg='#000000', # 黑色文字 bd=2, # 明确边框 relief=tk.SUNKEN, # 凹陷效果,明确输入区域 highlightthickness=1, # 轻微高亮 highlightbackground='#cccccc', wrap=tk.WORD, font=('Arial', 11) # 减小字体 ) self.input_text.grid(row=1, column=0, sticky=(tk.W, tk.E), pady=(0, 10)) # 发送按钮 send_button = tk.Button( dialog_container, text="发送", command=self.send_message, bg='#4a86e8', fg='white', relief=tk.FLAT, font=('Arial', 10, 'bold'), padx=20, pady=5 ) send_button.grid(row=2, column=0, sticky=tk.E) # 绑定回车键发送消息 self.input_text.bind('<Return>', self.on_enter_pressed) # 添加拖拽窗口的功能 self.root.bind("<Button-1>", self.start_move) self.root.bind("<B1-Motion>", self.do_move) def start_move(self, event): self.x = event.x self.y = event.y def do_move(self, event): deltax = event.x - self.x deltay = event.y - self.y x = self.root.winfo_x() + deltax y = self.root.winfo_y() + deltay self.root.geometry(f"+{x}+{y}") def on_enter_pressed(self, event): # 如果按下Shift+Enter,则换行;否则发送消息 if event.state & 0x1: # Shift键被按下 return # 让文本框正常换行 else: self.send_message() return "break" # 阻止文本框换行 def get_mcp_tool_list(self): """ 获取MCP服务器的可用工具列表(按需加载) """ self.load_mcp_tools_on_demand() # 确保工具列表已加载 return self.available_tools or {} # 返回已加载的数据或空字典 def show_all_methods(self): """ 显示所有可用的MCP工具 """ # 确保工具列表已加载 available_tools = self.get_mcp_tool_list() all_methods = "所有可用的MCP工具:\n\n" if available_tools: for server_name, tools in available_tools.items(): if tools: all_methods += f"服务器: {server_name}\n" for tool_name, tool_info in tools.items(): all_methods += f" - {tool_name}: {tool_info['description']}\n" all_methods += "\n" else: all_methods += f"服务器: {server_name} - 无可用工具\n\n" else: all_methods += "当前无可用的MCP工具\n" return all_methods def handle_run_command(self, method_number): """ 处理运行指定编号的方法 """ try: method_num = int(method_number.strip()) if method_num in self.method_map: method_info = self.method_map[method_num] # 根据方法类型生成相应的AI请求 if method_info['name'].startswith('调用'): # 这是一个MCP工具调用方法 tool_name = method_info['name'][2:] # 去掉'调用'前缀 return f"请调用{tool_name}方法" else: # 返回方法的描述 return f"执行{method_info['name']}: {method_info['description']}" else: return f"错误:方法编号 {method_num} 不存在" except ValueError: return f"错误:无效的方法编号 '{method_number}'" def enhance_prompt_with_tools(self, user_input): """ 将可用的MCP工具列表添加到用户输入中,增强提示词 """ # 获取可用的工具列表(这会触发首次加载) available_tools = self.get_mcp_tool_list() if available_tools: tool_description = "可用的MCP工具列表:\n" for server_name, tools in available_tools.items(): if tools: tool_names = [f"{name}({details['description']})" for name, details in tools.items()] tool_description += f"- {server_name}: {', '.join(tool_names)}\n" else: tool_description += f"- {server_name}: 无可用工具\n" enhanced_prompt = f""" {tool_description} 如果您需要调用MCP工具,请严格按照以下格式响应: [MCPCALL]server_name|tool_name|{{"param1": "value1", "param2": "value2"}}[/MCPCALL] 其中: - server_name: 服务器名称 - tool_name: 工具名称 - 参数部分必须是有效的JSON格式 如果不需要调用MCP工具,请正常回答。 用户请求: {user_input} """ return enhanced_prompt, True else: return user_input, False def parse_ai_response_for_mcp(self, ai_response): """ 解析AI响应,查找MCP工具调用指令 """ import re # 匹配 [MCPCALL]server_name|tool_name|{"params": "values"}[/MCPCALL] 格式 pattern = r'\[MCPCALL\](.*?)\|(.*?)\|({.*?})\[/MCPCALL\]' matches = re.findall(pattern, ai_response, re.DOTALL) if matches: server_name, tool_name, params_str = matches[0] try: arguments = json.loads(params_str) return True, server_name.strip(), tool_name.strip(), arguments except json.JSONDecodeError: print(f"参数解析失败: {params_str}") return False, None, None, None return False, None, None, None def send_message(self): # 获取输入内容 user_input = self.input_text.get(1.0, tk.END).strip() if not user_input: return # 检查是否为特殊命令 if user_input == '/h': help_text = self.show_all_methods() # 弹窗显示帮助信息而不是在窗口中显示 messagebox.showinfo("帮助", help_text) self.input_text.delete(1.0, tk.END) return # 检查是否为运行命令 (/r + 数字或普通文本) if user_input.startswith('/r ') and len(user_input) > 3: command_part = user_input[3:].strip() # 获取/r后面的部分 # 检查后面是否是纯数字 if command_part.isdigit(): # 是数字,按编号调用 ai_request = self.handle_run_command(command_part) # 显示用户输入 self.output_text.insert(tk.END, f">> {user_input}\n") # 处理AI请求 - 使用增强的提示词(带工具信息) thread = threading.Thread(target=self.process_message_with_tools, args=(ai_request,)) thread.start() self.input_text.delete(1.0, tk.END) return else: # 不是数字,作为普通消息发送,但使用增强的提示词(带工具信息) self.output_text.insert(tk.END, f">> {user_input}\n") thread = threading.Thread(target=self.process_message_with_tools, args=(user_input,)) thread.start() self.input_text.delete(1.0, tk.END) return # 显示用户输入(普通消息) self.output_text.insert(tk.END, f">> {user_input}\n") self.input_text.delete(1.0, tk.END) # 在新线程中处理普通消息(不带工具信息) thread = threading.Thread(target=self.process_message, args=(user_input,)) thread.start() def process_message(self, user_input): """处理普通消息(不添加工具信息到prompt)""" try: # 直接使用原始用户输入,不添加工具信息 messages = [{"role": "user", "content": user_input}] result = self.llm_service.create(messages) # 获取AI响应文本 response_text = json.dumps(result, indent=2, ensure_ascii=False) if isinstance(result, dict) else str(result) # 尝试解析AI响应中的MCP调用指令 is_mcp_call, server_name, tool_name, arguments = self.parse_ai_response_for_mcp(response_text) if is_mcp_call: # 执行MCP调用 self.root.after(0, lambda: self.output_text.insert(tk.END, f"正在调用 {server_name} 的 {tool_name} 工具...\n")) try: result = asyncio.run(self.mcp_client.call_tool(server_name, tool_name, arguments)) if result: # 处理MCP结果 result_content = [] for content in result.content: if hasattr(content, 'text'): result_content.append(content.text) elif hasattr(content, 'type'): result_content.append(f"[{content.type}]: {content}") result_text = "\n".join(result_content) if result_content else "无返回内容" # 将MCP结果展示在输出区域 self.root.after(0, lambda: self.output_text.insert(tk.END, f"MCP结果: {result_text}\n")) # 将MCP结果反馈给LLM进行进一步处理 feedback_prompt = f"用户请求: {user_input}\nMCP调用结果: {result_text}\n请根据这个结果向用户提供适当的回应。" llm_messages = [{"role": "user", "content": feedback_prompt}] llm_result = self.llm_service.create(llm_messages) # 提取并显示LLM响应中的content部分 llm_response = self.extract_content_from_response(llm_result) self.root.after(0, lambda r=llm_response: self.output_text.insert(tk.END, f"<< {r}\n\n")) else: self.root.after(0, lambda: self.output_text.insert(tk.END, "MCP调用未返回结果\n\n")) except Exception as e: error_msg = f"MCP调用失败: {str(e)}" self.root.after(0, lambda: self.output_text.insert(tk.END, f"{error_msg}\n\n")) else: # 如果AI响应不包含MCP调用,直接显示结果 # 提取并显示LLM响应中的content部分 llm_response = self.extract_content_from_response(result) # 检查原响应中是否包含用户输入的原始内容,避免重复显示 if "[MCPCALL]" not in response_text: self.root.after(0, lambda r=llm_response: self.output_text.insert(tk.END, f"<< {r}\n\n")) else: # 如果AI响应包含MCP调用格式但解析失败,也显示结果 self.root.after(0, lambda r=llm_response: self.output_text.insert(tk.END, f"<< {r}\n\n")) except Exception as e: self.root.after(0, lambda: self.output_text.insert(tk.END, f"错误: {str(e)}\n")) def process_message_with_tools(self, user_input): """处理消息(添加工具信息到prompt)""" try: # 使用增强的提示词获取AI响应(添加工具信息) enhanced_prompt, has_tools = self.enhance_prompt_with_tools(user_input) if has_tools: # 将工具列表信息提供给AI messages = [{"role": "user", "content": enhanced_prompt}] else: messages = [{"role": "user", "content": user_input}] result = self.llm_service.create(messages) # 获取AI响应文本 response_text = json.dumps(result, indent=2, ensure_ascii=False) if isinstance(result, dict) else str(result) # 尝试解析AI响应中的MCP调用指令 is_mcp_call, server_name, tool_name, arguments = self.parse_ai_response_for_mcp(response_text) if is_mcp_call: # 执行MCP调用 self.root.after(0, lambda: self.output_text.insert(tk.END, f"正在调用 {server_name} 的 {tool_name} 工具...\n")) try: result = asyncio.run(self.mcp_client.call_tool(server_name, tool_name, arguments)) if result: # 处理MCP结果 result_content = [] for content in result.content: if hasattr(content, 'text'): result_content.append(content.text) elif hasattr(content, 'type'): result_content.append(f"[{content.type}]: {content}") result_text = "\n".join(result_content) if result_content else "无返回内容" # 将MCP结果展示在输出区域 self.root.after(0, lambda: self.output_text.insert(tk.END, f"MCP结果: {result_text}\n")) # 将MCP结果反馈给LLM进行进一步处理 feedback_prompt = f"用户请求: {user_input}\nMCP调用结果: {result_text}\n请根据这个结果向用户提供适当的回应。" llm_messages = [{"role": "user", "content": feedback_prompt}] llm_result = self.llm_service.create(llm_messages) # 提取并显示LLM响应中的content部分 llm_response = self.extract_content_from_response(llm_result) self.root.after(0, lambda r=llm_response: self.output_text.insert(tk.END, f"<< {r}\n\n")) else: self.root.after(0, lambda: self.output_text.insert(tk.END, "MCP调用未返回结果\n\n")) except Exception as e: error_msg = f"MCP调用失败: {str(e)}" self.root.after(0, lambda: self.output_text.insert(tk.END, f"{error_msg}\n\n")) else: # 如果AI响应不包含MCP调用,直接显示结果 # 提取并显示LLM响应中的content部分 llm_response = self.extract_content_from_response(result) # 检查原响应中是否包含用户输入的原始内容,避免重复显示 if "[MCPCALL]" not in response_text: self.root.after(0, lambda r=llm_response: self.output_text.insert(tk.END, f"<< {r}\n\n")) else: # 如果AI响应包含MCP调用格式但解析失败,也显示结果 self.root.after(0, lambda r=llm_response: self.output_text.insert(tk.END, f"<< {r}\n\n")) except Exception as e: self.root.after(0, lambda: self.output_text.insert(tk.END, f"错误: {str(e)}\n")) def extract_content_from_response(self, response): """ 从LLM响应中提取content部分 """ if isinstance(response, dict): # 如果response是字典,尝试从中提取content choices = response.get('choices', []) if choices and len(choices) > 0: message = choices[0].get('message', {}) content = message.get('content', '') if content: return content # 如果没有找到content,返回整个响应的字符串表示 return json.dumps(response, indent=2, ensure_ascii=False) else: # 如果response不是字典,直接返回字符串形式 return str(response) def display_result(self, result): try: response_text = self.extract_content_from_response(result) self.output_text.insert(tk.END, f"<< {response_text}\n\n") self.output_text.see(tk.END) # 滚动到底部 except Exception as e: self.output_text.insert(tk.END, f"<< 处理结果时出错: {str(e)}\n\n") def main(): root = tk.Tk() # 设置窗口属性 root.wm_attributes("-topmost", 1) # 置顶 root.attributes('-alpha', 0.98) # 设置更高透明度 app = MCPAICaller(root) root.mainloop() if __name__ == "__main__": main()

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/1189526.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

“人类心智制定和解决复杂问题的能力,与在现实世界中实现客观理性行为所需处理的问题规模相比,显得极为有限。” — 赫伯特·A·西蒙(Herbert A. Simon)

“人类心智制定和解决复杂问题的能力,与在现实世界中实现客观理性行为所需处理的问题规模相比,显得极为有限。” — 赫伯特A西蒙(Herbert A. Simon) 文章目录 “人类心智制定和解决复杂问题的能力,与在现实世界中实现客观理性行为所需处理的问题规模相比,显得极为有限。”…

C++算法训练第八天

C++算法训练第八天 以下为牛客挑战 今日收获 学习到了ksm的写法 int ksm(int p,int q,int mod){int result=1;p=p%mod;while (q>0){if(q&1){//result=(1ll*result*p)%mod;}q=q>>1;p=(1ll*p*p)%mod;}retur…

2026年上海防水补漏行业顶尖企业权威评测:全面解析防水、修复、翻新与检测服务 - shruisheng

随着上海城市建筑步入成熟期,各类建筑渗漏问题日益成为影响居住安全与建筑寿命的关键因素。为帮助业主及管理方精准选择可靠的防水服务商,本次评测严格依据上海建筑防水行业协会发布的2025年度行业数据,结合企业技术…

【实战项目】 方正科技市场营销策略研究

运行效果:https://lunwen.yeel.cn/view.php?id=5841 方正科技市场营销策略研究摘要:本文以方正科技为研究对象,对其市场营销策略进行了深入分析。通过文献研究、市场调研和案例研究等方法,探讨了方正科技的市场定…

【Web安全】SSRF - 教程

pre { white-space: pre !important; word-wrap: normal !important; overflow-x: auto !important; display: block !important; font-family: "Consolas", "Monaco", "Courier New", …

聊聊五种 Redis 部署模式

这篇文章&#xff0c;分享自己职业生涯经历的五种 Redis 部署模式&#xff0c;希望对大家有所启发。1 单实例这是 Redis 最简单、最基础的部署方式&#xff0c;即&#xff1a;整个 Redis 服务运行在单个服务器和单个进程中。笔者第一次在生产环境使用 Redis &#xff0c;是在艺…

京东e卡回收真的靠谱吗?揭秘背后真相! - 京顺回收

不少人手中积压着单位发的京东e卡,想变现却顾虑重重。网上“高价回收”的平台良莠不齐,和陌生人交易又担心卡密泄露。据行业报告,2025年国内闲置京东e卡规模超200亿元,六成用户因流程不熟而遭遇诈骗。那么,如何判…

[MCP] Prompt

Sometimes there are common workflows for people using your MCP server you want to make easier for users. You may not want your users to have to write the same prompt all the time for that workflow (not…

从复杂到有序:汽车制造企业多元数据库管理走向自治智能的实践观察

在追求“极致效率”的汽车制造领域&#xff0c;每一秒钟的生产节拍都依赖于底层数据流的平稳支撑。本案例客户是国内某知名合资车企&#xff0c;拥有年产百万辆规模的行业巨头&#xff0c;其数字化版图已延伸至研发、供应链、生产制造及营销全链路。支撑这套庞大体系的&#xf…

写论文软件哪个好?实测科普!宏智树 AI 凭 “学术真功夫” 成毕业刚需

作为深耕论文写作科普的教育测评博主&#xff0c;每年毕业季后台都会被 “写论文软件哪个好” 的提问刷屏。对比十余款工具后发现&#xff0c;多数软件要么是 “文字拼接器”&#xff0c;生成内容空洞无物&#xff1b;要么存在虚假引用、数据失真的隐患&#xff0c;踩中了学术规…

2026年硅胶模具厂家深度选型指南:食品级与医用级需求下的三大方案解析 - 博客万

引言 在食品加工、医疗设备等领域,硅胶模具的选择直接影响产品质量与生产效率。食品厂关心模具的食品级认证、成型精准度与脱模效率,医疗设备厂注重生物相容性与质量稳定性,中小商家则关注性价比与交付速度。面对市…

【实战项目】 基于springboot的前后端分离学生健康体检管理系统

运行效果:https://lunwen.yeel.cn/view.php?id=5848 基于springboot的前后端分离学生健康体检管理系统摘要:随着我国教育事业的不断发展,学生健康体检管理工作的重要性日益凸显。为提高学生健康体检管理的效率和质…

当 Agent 进入系统阶段,AI 产品开始真正分化

&#x1f44b;前言过去一年&#xff0c;AI 产品讨论里有两个词被反复提及&#xff1a;模型&#xff0c;和 Agent。但在越来越多真实项目中&#xff0c;一个更现实的问题正在浮现&#xff1a;当 Agent 不再是单点能力&#xff0c;而开始参与业务流程、并发任务和协同决策时&…

2026年知名的公务车品牌厂家推荐及行业发展解析 - 品牌排行榜

公务车品牌厂家在城市公共服务、商业运营等领域发挥着重要作用,其产品性能、可靠性与适用性直接关系到服务效率与运营成本。随着新能源技术的发展和应用场景的细分,市场对公务车的需求正朝着智能化、定制化、高效能方…

【实战项目】 数字孪生在水利调度中的应用

运行效果:https://lunwen.yeel.cn/view.php?id=5829 数字孪生在水利调度中的应用摘要:随着信息技术的发展,数字孪生技术在各个领域的应用日益广泛。本文针对水利调度领域,研究了数字孪生技术的应用,旨在提高水利…

2026Q1唐山口碑财税公司推荐榜:正规备案为基 - 品牌智鉴榜

2026Q1,唐山财税市场迎来政策细化与监管升级的双重格局监管步入精准化阶段,钢铁、装备制造等主导产业面临需精准对接技改税收抵扣优化需求政策,初创企业则亟需合规高效的财税支撑面临合规记账、高效办证的双重需求,…

【实战项目】 基于单片机激光打靶语音播报系统的设计与实现

运行效果:https://lunwen.yeel.cn/view.php?id=5842 基于单片机激光打靶语音播报系统的设计与实现摘要:本论文针对传统激光打靶训练中信息反馈单一的问题,设计并实现了一套基于单片机的激光打靶语音播报系统。系统…

【RPA】拼多多商家后台取数口径

拼多多RPA取数口径 后台主页发货管理 订单查询 可检索全部订单、待付款、待发货、待收货、退款/售后等订单。物流概况商品管理 商品列表左侧菜单中选择【商品管理】,点击【商品列表】,即可看到当前店铺所有商品信息。…

9 款 AI 写论文哪个好?实测封神!宏智树 AI 凭硬核实力稳坐头把交椅

毕业论文写作堪称学术生涯的 “终极闯关”&#xff0c;选题迷茫、文献堆砌、数据失真、查重超标&#xff0c;每一个环节都能让人焦头烂额。作为深耕论文写作科普的测评博主&#xff0c;我耗时两周&#xff0c;围绕文献真实性、功能全面性、数据可视化、学术合规性四大核心维度&…

【实战项目】 基于DSP新型电能质量监测装置的研究

运行效果:https://lunwen.yeel.cn/view.php?id=5827 基于DSP新型电能质量监测装置的研究摘要:本文针对当前电能质量监测领域的技术需求,研究了基于DSP(数字信号处理器)的新型电能质量监测装置的设计与实现。通过…