吴恩达MCP课程(4):connect_server_mcp_chatbot

article/2025/7/20 10:16:33

目录

    • 完整代码
    • 代码解释
      • 1. 导入和初始化
      • 2. 类型定义
      • 3. MCP_ChatBot 类初始化
      • 4. 查询处理 (process_query)
      • 5. 服务器连接管理
      • 6. 核心特性总结
    • 示例

完整代码

原课程代码是用Anthropic写的,下面代码是用OpenAI改写的,模型则用阿里巴巴的模型做测试
.env 文件为:

OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
OPENAI_API_BASE=https://dashscope.aliyuncs.com/compatible-mode/v1

另外,课程代码只是单轮对话,下面代码修改为多轮对话,更适合千问模型的调用方式

from dotenv import load_dotenv
import openai
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from typing import List,TypedDict
from contextlib import AsyncExitStack
from typing import Dict
import asyncio
import json
import osload_dotenv()class ToolDefinition(TypedDict):name: strdescription: strinput_schema: dictclass MCP_ChatBot:def __init__(self):# Initialize session and client objectsself.sessions: List[ClientSession] = []  # newself.exit_stack = AsyncExitStack()self.client = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"),base_url=os.getenv("OPENAI_API_BASE"))self.available_tools: List[ToolDefinition] = []  # newself.tool_to_session: Dict[str, ClientSession] = {}self.messages = []async def process_query(self, query):self.messages.append({'role':'user', 'content':query})response = self.client.chat.completions.create(model='qwen-turbo',# max_tokens=2024,tools=self.available_tools,messages=self.messages )process_query = Truewhile process_query:# 获取助手的回复message = response.choices[0].message# 检查是否有普通文本内容if message.content:print(message.content)process_query = False# 检查是否有工具调用elif message.tool_calls:# 添加助手消息到历史self.messages.append({"role": "assistant", "content": None,"tool_calls": message.tool_calls})# 处理每个工具调用for tool_call in message.tool_calls:tool_id = tool_call.idtool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)print(f"Calling tool {tool_name} with args {tool_args}")# 执行工具调用session = self.tool_to_session[tool_name]result = await session.call_tool(tool_name, arguments=tool_args)# 添加工具结果到消息历史self.messages.append({"role": "tool","tool_call_id": tool_id,"content": result.content})# 获取下一个回复response = self.client.chat.completions.create(model='qwen-turbo',# max_tokens=2024,tools=self.available_tools,messages=self.messages )self.messages.append({"role": "assistant", "content": response.choices[0].message.content})# 如果只有文本回复,则结束处理if response.choices[0].message.content and not response.choices[0].message.tool_calls:print(response.choices[0].message.content)process_query = Falseasync def chat_loop(self):"""Run an interactive chat loop"""print("\nMCP Chatbot Started!")print("Type your queries or 'quit' to exit.")while True:try:query = input("\nQuery: ").strip()if query.lower() == 'quit':breakawait self.process_query(query)print("\n")except Exception as e:print(f"\nError: {str(e)}")async def connect_to_server(self, server_name: str, server_config: dict) -> None:"""Connect to a single MCP server."""try:server_params = StdioServerParameters(**server_config)stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))read, write = stdio_transportsession = await self.exit_stack.enter_async_context(ClientSession(read, write))await session.initialize()self.sessions.append(session)# List available tools for this sessionresponse = await session.list_tools()tools = response.toolsprint(f"\nConnected to {server_name} with tools:", [t.name for t in tools])for tool in tools:  # newself.tool_to_session[tool.name] = sessionself.available_tools.append({"type": "function","function": {"name": tool.name,"description": tool.description,"parameters": tool.inputSchema}})except Exception as e:print(f"Failed to connect to {server_name}: {e}")async def connect_to_servers(self):  # new"""Connect to all configured MCP servers."""try:with open("server_config.json", "r") as file:data = json.load(file)servers = data.get("mcpServers", {})for server_name, server_config in servers.items():await self.connect_to_server(server_name, server_config)except Exception as e:print(f"Error loading server configuration: {e}")raise           async def clenup(self):await self.exit_stack.aclose()async def main():chatbot = MCP_ChatBot()try:await chatbot.connect_to_servers()await chatbot.chat_loop()finally:await chatbot.clenup()if __name__ == "__main__":asyncio.run(main())"""
1、Fetch the content of this website: https://modelcontextprotocol.io/docs/concepts/architecture. 
2、save the content in the file "mcp_summary.md"
"""

代码解释

1. 导入和初始化

from dotenv import load_dotenv
import openai
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from typing import List,TypedDict
from contextlib import AsyncExitStack
from typing import Dict
import asyncio
import json
import osload_dotenv()
  • 导入必要的库,包括OpenAI客户端、MCP协议相关模块、异步处理模块等
  • load_dotenv() 加载环境变量配置

2. 类型定义

class ToolDefinition(TypedDict):name: strdescription: strinput_schema: dict

定义工具的类型结构,用于类型提示。

3. MCP_ChatBot 类初始化

class MCP_ChatBot:def __init__(self):self.sessions: List[ClientSession] = []  # 存储多个MCP会话self.exit_stack = AsyncExitStack()  # 管理异步资源self.client = openai.OpenAI(  # OpenAI客户端api_key=os.getenv("OPENAI_API_KEY"),base_url=os.getenv("OPENAI_API_BASE"))self.available_tools: List[ToolDefinition] = []  # 可用工具列表self.tool_to_session: Dict[str, ClientSession] = {}  # 工具名到会话的映射self.messages = []  # 对话历史

关键特性:

  • 多会话支持sessions 列表存储多个MCP服务器会话
  • 工具映射tool_to_session 将工具名映射到对应的会话,实现工具路由
  • 资源管理:使用 AsyncExitStack 管理异步资源的生命周期

4. 查询处理 (process_query)

async def process_query(self, query):self.messages.append({'role':'user', 'content':query})response = self.client.chat.completions.create(model='qwen-turbo',tools=self.available_tools,messages=self.messages )

核心处理逻辑:

  1. 消息循环处理:使用 while process_query 循环处理多轮对话
  2. 工具调用处理:检测并执行工具调用,通过 tool_to_session 路由到正确的MCP服务器
  3. 结果整合:将工具执行结果添加到对话历史中

5. 服务器连接管理

async def connect_to_server(self, server_name: str, server_config: dict) -> None:# 建立单个服务器连接server_params = StdioServerParameters(**server_config)stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))# ... 获取工具并建立映射for tool in tools:self.tool_to_session[tool.name] = sessionself.available_tools.append({"type": "function","function": {"name": tool.name,"description": tool.description,"parameters": tool.inputSchema}})
async def connect_to_servers(self):# 连接所有配置的服务器with open("server_config.json", "r") as file:data = json.load(file)servers = data.get("mcpServers", {})for server_name, server_config in servers.items():await self.connect_to_server(server_name, server_config)

6. 核心特性总结

多服务器支持

  • 可以同时连接多个MCP服务器
  • 每个服务器的工具都被统一管理
  • 通过工具名自动路由到正确的服务器

OpenAI格式兼容

  • 工具定义使用OpenAI的函数调用格式
  • 支持完整的工具调用流程

异步处理

  • 全异步设计,支持并发处理
  • 使用 AsyncExitStack 管理资源生命周期

配置化管理

  • 通过 server_config.json 配置多个服务器
  • 支持动态加载服务器配置

这个实现相比单服务器版本的主要优势是可以整合多个不同功能的MCP服务器,为用户提供更丰富的工具集合。

示例

uv run connect_server_map_chatbot.py

请添加图片描述


http://www.hkcw.cn/article/HQTkbPRZwn.shtml

相关文章

网页前端开发(基础进阶1--盒子模型)

颜色表示方法3种: 1.关键字: color:green; gray red yellow 2.rgb表示法:红,绿,蓝三原色。rgb(r,g,b),r表示红色,g表示绿…

中国造船是如何取代韩国霸主地位的 14年激战成就全球王者

在全球造船业的百年激荡中,主导权的更迭总是伴随着技术、规模与国家意志的碰撞。从19世纪的英国,到战后的日本,再到20世纪末的韩国,造船业王冠几度易主。如今,中国以雷霆万钧之势终结了韩国长达20年的霸主地位。2024年,中国造船业以74.7%的新船订单份额,全面改写全球造船…

加沙城四分之三水井被以军摧毁 缺水危机加剧

加沙城政府部门6月1日发布消息称,由于以军摧毁了当地约75%的水井,且剩余水井缺乏正常运行所需燃料,加沙城正面临严重的缺水问题和灾难性危机。该部门警告称,如果国际社会和相关机构不尽快介入,可能会引发严重的健康和环境危机。责任编辑:zhangxiaohua

马克龙接见巴黎圣日耳曼全队 谴责暴力事件

巴黎圣日耳曼队在5月31日晚赢得了欧冠联赛冠军奖杯,法国多地的球迷彻夜庆祝。然而,庆祝活动期间发生了多起暴力事件。据法国内政部消息,狂欢夜有559人因滋事被捕,并发生了两起命案。法国总统马克龙6月1日在爱丽舍宫接见球队时,严厉谴责了这些暴力事件,称其“不可接受”。…

C++实现汉诺塔游戏自动完成

目录 一、汉诺塔的规则二、数学递归推导式三、步骤实现(一)汉诺塔模型(二)递归实现(三)显示1.命令行显示2.SDL图形显示 四、处理用户输入及SDL环境配置五、总结六、源码下载 一、汉诺塔的规则 游戏由3根柱子和若干大小不一的圆盘组成,初始状态下,所有的…

抖音评论数据采集解决方案推荐

内容概要 在当今数字化营销的背景下,抖音作为一款广受欢迎的短视频平台,吸引了众多用户和企业的关注。为了更好地理解用户反馈和优化营销策略,评论数据的采集显得尤为重要。本文将为读者提供一系列系统化的解决方案,涵盖多款高效…

若依框架-代码生成器的使用

1.服务启动 注意一定要启动Gen\GateWay\System 2.创建表 注意一定要写上字段的注释和表的注释。 1.注意主键的命名 2.注意驼峰命名法 3.代码生成 3.1 按照以下箭头步骤,选择生成的表,点击确定 3.2 核对生成表的结构 3.3 编辑生成表的基础信息 3.4 点…

河南46岁卡车司机已下葬 卡友团队千里送别

近日,46岁的河南卡车司机常志荣在青藏高原离世,卡友团队一同将其骨灰接回老家。6月1日,常志荣已经在老家安葬。他去世后留下了重组家庭的6个孩子,其中5个孩子还在上学。车友任先生透露,常志荣出发前同行曾建议他至少携带两罐氧气,但他为了省下30元费用,最终只购买了一罐…

交警回应豪车车主赖停车费被举报酒驾 代驾小哥报警揭发真相

近日,浙江宁波一名兰博基尼车主酒后请代驾将车开到小区,随后自己开车进车库。因不愿支付代驾小哥垫付的5元停车费,被对方举报酒驾。经检测,该车主血液中的酒精含量达59mg/100ml,达到酒驾标准,其驾驶证被扣12分,暂扣6个月,并被处以2000元以下罚款。代驾小哥石先生表示,…

Nat. Commun|借助深度学习分层方法,量化胎盘组织学全切片图像中细胞和显微解剖组织结构的变异性

小罗碎碎念 这篇文章提出了用于胎盘组织学全切片图像分析的深度学习框架HAPPY,其核心是通过三阶段流水线实现从细胞到组织结构的层次化建模。 首先利用RetinaNet定位细胞核。 再通过ResNet-50分类11种细胞类型 最后基于ClusterGCN图神经网络将细胞聚合成9类显微组织…

融智学道函数智慧系统研习(图文并茂)

融智学道函数智慧系统研习(图文并茂) 融智学创立者邹晓辉亲自设计和审订并规划布局人机互助制作全过程 摘要:本文呈现邹晓辉融智学理论的核心架构,包含四大公式可视化系统:1)道函数(f_Tao0&am…

小米YU7将推出5款全新车身颜色 更多经典配色即将揭晓

6月1日的消息显示,在5月22日的发布会上,小米YU7公布了四款颜色:钛金属色、宝石绿、熔岩橙和寒武岩灰。官方透露,除了已发布的四款颜色外,还有五款新颜色即将发布。回顾小米SU7刚推出时,提供了9种颜色选择,分为跑车色系、时尚色系、豪华色系和经典色系。基础款中的海湾蓝…

国足出征印尼25人大名单 刘若钒因伤落选

北京时间6月2日,中国男足公布了出征印度尼西亚雅加达的25人名单。刘若钒因伤落选,他在国足上海集训期间受到伤病困扰,缺席了数次合练。尽管恢复了有球训练,但最终未能进入出征印尼的名单。谢文能和林良铭将在对阵印度尼西亚的比赛里停赛。不过,两人还是会随队前往雅加达,…

国防大学副校长香会发言释放何信号 坚定维护亚太和平稳定

今年在新加坡举行的香格里拉对话会上,最重要的发言来自中国国防大学代表团团长、国防大学副校长兼教育长胡钢锋少将。他提出了“三个坚持三个反对”的倡议,这不仅反映了中国对亚太地区海上安全合作的看法,也展示了中国维护和平与稳定的决心。当地时间5月31日下午,胡钢锋在香…

尾号0000000的手机号拍出61.2万元 司法拍卖市场火热

端午节期间,司法拍卖市场异常活跃。6月1日,一个尾号为七个零的手机号码使用权以25万元起拍,吸引了13人报名参与竞拍。经过激烈竞争,该号码最终以61.2万元成交。据竞买公告显示,截至2025年4月23日,该手机号码无欠费,余额约9.14元。套餐为4G全国流量王8元套餐,未绑定宽带…

上位机系统 设备初始化Load界面开发

效果 设备加载完后打开主界面 思路 1.创建一个 SplashWindow.xaml 就是上面效果图,主要是显示系统名称和正在加载的信息,进度条。 2.在 SplashWindow 的后台执行加载代码。 3.最后打开主界面 为什么要写这片文章,以前的项目也有这个界面但实现都很复杂,用到了线程间的数…

SQL Relational Algebra(数据库关系代数)

目录 What is an “Algebra” What is Relational Algebra? Core Relational Algebra Selection Projection Extended Projection Product(笛卡尔积) Theta-Join Natural Join Renaming Building Complex Expressions Sequences of Assignm…

董宇辉龙舟夺冠又赢鸭 传统文化焕发新活力

5月31日,陕西安康迎来了第25届汉江龙舟节的开幕。开幕式上,龙舟方阵展演、龙舟横渡汉江以及抢鸭子、摸鲤鱼等传统环节轮番上演,节日氛围浓厚。据了解,今年有27支队伍共600多名选手参加了龙舟竞渡比赛。当天,董宇辉现身安康龙舟文化园,与现场市民和游客热情互动,并齐声高…

为何大巴黎没了姆巴佩进球反而更多 全攻全守战术奏效

巴黎圣日耳曼夺得队史第一座欧冠冠军奖杯,超级巨星姆巴佩在社交账号上向老东家表示祝贺:“伟大的日子终于来临,这场胜利是属于整个俱乐部的荣光,恭喜巴黎圣日耳曼!”过去六个赛季,姆巴佩以大巴黎当家球星的身份夺得法甲金靴奖。他去年夏天从大巴黎加盟上赛季欧冠冠军皇家…

俄称击落162架乌军无人机 俄军持续攻势

当地时间5月30日,俄罗斯国防部发布战报称,在过去一周里,俄军对乌克兰境内的国防工业设施、军用机场基础设施、武器弹药储存设施等目标实施了打击。在苏梅、哈尔科夫、顿涅茨克等地,俄军打退了乌军多次进攻并发动攻势。俄防空部队击落了1439架固定翼无人机,并控制了苏梅、哈…