AI客服系统如何实现灵活扩展？MCP协议+Skills体系深度解析

本章目录MCP 是什么为什么重要实现一个简单的 MCP ServerMCP Client 调用工具Skills 技能包设计技能包动态加载MCP Skills 完整集成MCP 是什么为什么重要MCPModel Context Protocol是 Anthropic 在 2024 年底推出的开放标准。一句话解释AI 领域的 USB 接口。没有 MCP 的时候接一个新工具 → 写适配代码 → 测试 → 部署再来一个新工具 → 再写适配代码 → 再测试 → 再部署...无限循环有了 MCP工具提供方实现一次 MCP Server → 所有支持 MCP 的 AI 应用直接用架构┌──────────────────────┐│ AI 应用MCP Host │ ← 咱们的客服 Agent└──────────┬───────────┘ │ MCP 协议JSON-RPC over stdio/SSE┌──────────▼───────────┐│ MCP Client │ ← 内置在 AI 应用里└──────────┬───────────┘ │ ┌──────┼──────┬──────────┐ ▼ ▼ ▼ ▼┌──────┐┌──────┐┌──────┐┌──────────┐│订单 ││物流 ││CRM ││外部 API ││Server││Server││Server││Server │└──────┘└──────┘└──────┘└──────────┘对咱们的项目意味着什么查订单、查物流、查库存这些工具不用硬编码在 Agent 代码里各个业务团队可以独立开发自己的 MCP Server新接一个业务系统只需要启动对应的 MCP ServerAgent 自动就能用实现一个简单的 MCP Server文件01_mcp_server.pyMCP Server 本质上是一个实现了 MCP 协议的 HTTP/WebSocket 服务。本章用简化版演示核心概念import jsonclass SimpleMCPServer: 简化的 MCP Server 实现。 真实的 MCP Server 需要 - 实现 JSON-RPC 2.0 协议 - 支持 tools/list、tools/call 等 MCP 标准方法 - 通过 stdio 或 HTTP/SSE 通信 - 支持资源resources、提示prompts等扩展 这里用简化版展示核心逻辑工具注册 请求处理。 def __init__(self, name: str, description: str): self.name name self.description description self.tools {} # tool_name → {name, description, handler, parameters} def register_tool(self, name: str, description: str, handler: callable, parameters: dict None): 注册一个工具。 self.tools[name] { name: name, description: description, handler: handler, parameters: parameters or {}, } print(f [MCP Server:{self.name}] 注册工具: {name}) def list_tools(self) - list: 列出所有已注册的工具对应 MCP 协议的 tools/list 方法。 return [ { name: t[name], description: t[description], parameters: t[parameters], server: self.name, } for t in self.tools.values() ] def call_tool(self, tool_name: str, arguments: dict None) - str: 调用指定工具对应 MCP 协议的 tools/call 方法。 arguments arguments or {} tool self.tools.get(tool_name) if not tool: return json.dumps({error: f工具 {tool_name} 不存在}, ensure_asciiFalse) try: result tool[handler](**arguments) if isinstance(result, dict): return json.dumps(result, ensure_asciiFalse) return str(result) except TypeError as e: return json.dumps({error: f参数错误: {e}}, ensure_asciiFalse) except Exception as e: return json.dumps({error: f执行错误: {e}}, ensure_asciiFalse) def handle_request(self, method: str, params: dict None) - dict: 处理 MCP 请求模拟 JSON-RPC 协议。 params params or {} if method tools/list: return {tools: self.list_tools()} elif method tools/call: tool_name params.get(name) arguments params.get(arguments, {}) result self.call_tool(tool_name, arguments) # 标准 MCP 响应格式content 数组 return {content: [{type: text, text: result}]} else: return {error: f未知方法: {method}}使用示例——创建一个订单查询 MCP Serverorder_server SimpleMCPServer(order_service, 订单管理服务)MOCK_ORDERS { ORD-20240125-001: {product: StarPods Pro, status: shipped, price: 899},}order_server.register_tool( query_order, 查询订单详情, lambda order_id: MOCK_ORDERS.get(order_id, {error: 未找到}), {type: object, properties: {order_id: {type: string}}},)# tools/list让 Client 知道有哪些工具tools order_server.handle_request(tools/list)# → {tools: [{name: query_order, description: 查询订单详情, ...}]}# tools/call调用具体工具result order_server.handle_request(tools/call, { name: query_order, arguments: {order_id: ORD-20240125-001},})# → {content: [{type: text, text: {product: StarPods Pro, status: shipped}}]}MCP Client 调用工具文件02_mcp_client.pyMCP Client 的作用是连接 MCP Server发现可用工具调用工具。class SimpleMCPClient: 简化的 MCP Client。 职责 1. 连接多个 MCP Server 2. 发现所有 Server 上的可用工具 3. 统一工具调用接口不用关心工具在哪个 Server 上 def __init__(self): self.servers {} # name → SimpleMCPServer def connect(self, server: SimpleMCPServer): 连接一个 MCP Server。 self.servers[server.name] server print(f [MCP Client] 已连接 Server: {server.name} ({len(server.tools)} 个工具)) def disconnect(self, server_name: str): 断开与指定 Server 的连接。 if server_name in self.servers: del self.servers[server_name] def discover_tools(self) - list: 从所有已连接的 Server 发现可用工具。 all_tools [] for server in self.servers.values(): all_tools.extend(server.list_tools()) return all_tools def call_tool(self, tool_name: str, arguments: dict None) - str: 调用指定工具自动查找工具所在的 Server。 Agent 调用工具时只需要知道工具名 Client 自动路由到正确的 Server。 for server in self.servers.values(): if tool_name in server.tools: print(f [MCP Client] 调用 {server.name}.{tool_name}) return server.call_tool(tool_name, arguments) return json.dumps({error: f未找到工具 {tool_name}}, ensure_asciiFalse) def status(self) - dict: 返回 Client 状态摘要。 return { connected_servers: len(self.servers), server_names: list(self.servers.keys()), total_tools: len(self.discover_tools()), }把两个 Server 接进来Agent 只需跟 Client 交互# 两个 MCP Server各自负责一个业务域order_server SimpleMCPServer(order_service, 订单管理服务)product_server SimpleMCPServer(product_service, 产品信息服务)# Client 一次性连接两个 Serverclient SimpleMCPClient()client.connect(order_server) # [MCP Client] 已连接 Server: order_service (2 个工具)client.connect(product_server) # [MCP Client] 已连接 Server: product_service (2 个工具)# 发现全部工具——Agent 看到的是统一的工具列表tools client.discover_tools()# → [query_order, query_logistics, query_inventory, check_refund_policy]# 调用工具——Client 自动路由到正确的 Serverresult client.call_tool(query_order, {order_id: ORD-001})# [MCP Client] 调用 order_service.query_order# 新增一个 Server只需 connect() 一次Agent 代码不用改Skills 技能包设计文件03_skills.py如果说 MCP 解决的是怎么连通信协议那 Skills 解决的就是怎么管能力组织。一个 Skill 是一组相关能力的打包class Skill: 技能包 —— 把相关的 Tools Prompt 知识打包在一起。 类比 - MCP Server 像是工具箱 - Skill 像是技能书——告诉你面对什么问题时该用什么工具、怎么用 Skill 和 Multi-Agent 的关系 - Multi-Agent 是人的分工不同 Agent 扮演不同角色 - Skills 是能力的组织同一个 Agent 可以加载不同技能包 def __init__(self, name: str, description: str, system_prompt: str , required_tools: list None, required_servers: list None, knowledge: str , examples: list None): self.name name self.description description self.system_prompt system_prompt self.required_tools required_tools or [] self.required_servers required_servers or [] self.knowledge knowledge self.examples examples or [] def get_full_prompt(self, tool_descriptions: str ) - str: 获取完整的系统 Prompt含工具描述和知识文档。 parts [self.system_prompt] if tool_descriptions: parts.append(f\n\n可用工具\n{tool_descriptions}) if self.knowledge: parts.append(f\n\n知识库\n{self.knowledge}) if self.examples: parts.append(\n\n参考示例) for ex in self.examples: parts.append(f 用户{ex[user]}\n 回复{ex[assistant]}) return .join(parts)客服场景的三个技能包——每个包声明自己需要哪些工具和 Server# 退款技能包refund_skill Skill( namerefund, description处理退款、退货、换货请求, system_prompt你是星辰科技的退款专员。处理流程1. 查询订单 2. 判断退货条件 3. 告知操作步骤。, required_tools[query_order, check_refund_policy], required_servers[order_service, product_service], knowledge退款政策- 7天无理由退货签收后 7 天内可无理由退换- 15天质量问题质量问题 15 天内免费换新- 退款方式原路退回3-5 个工作日到账, examples[ {user: 我买了 3 天能退吗, assistant: 您购买 3 天完全在 7 天无理由退货期内请在 APP「我的订单」申请退款即可。}, ],)# 物流查询技能包logistics_skill Skill( namelogistics, description查询物流和订单配送状态, system_prompt你是星辰科技的物流查询专员。处理流程1. 查订单 2. 查物流 3. 告知详情。, required_tools[query_order, query_logistics], required_servers[order_service],)# 技术支持技能包tech_skill Skill( nametech_support, description处理产品故障和使用咨询, system_prompt你是星辰科技的技术支持工程师。耐心专业避免过于技术化的术语。, required_tools[query_order, query_inventory], required_servers[order_service, product_service], knowledge常见问题排查1. 蓝牙连不上确认开启 → 忘记配对 → 重新配对2. 有杂音清洁触点 → 换设备测试 → 重置耳机3. 充不进电换线 → 清洁充电盒触点 → 检查进水,)跟 Multi-Agent 的关系Multi-Agent 是人的分工不同 Agent 扮演不同角色Skills 是能力的组织同一个 Agent 可以加载不同技能包两者可以结合每个 Agent 默认加载自己的专属技能包需要时也可以借用其他技能包技能包动态加载文件04_skill_manager.pySkillManager 负责根据用户问题自动选择最合适的技能包加载class SkillManager: 技能包管理器 —— 根据用户问题动态选择技能包。 def __init__(self): self.skills {} # name → Skill self.mcp_client SimpleMCPClient() def register_skill(self, skill: Skill): self.skills[skill.name] skill def register_server(self, server: SimpleMCPServer): self.mcp_client.connect(server) def select_skill(self, user_message: str) - Optional[Skill]: 根据用户消息选择最合适的技能包。 llm get_llm(temperature0) skill_list \n.join( f- {name}: {skill.description} for name, skill in self.skills.items() ) prompt f根据用户消息选择最合适的技能包。可用技能包{skill_list}请只输出技能包名称不要输出其他内容。 response llm.invoke([HumanMessage(contentprompt)]) skill_name response.content.strip().lower() # 模糊匹配 for name in self.skills: if name in skill_name or skill_name in name: return self.skills[name] return list(self.skills.values())[0] def get_skill_tools(self, skill: Skill) - list: 获取技能包可用的工具列表。 available self.mcp_client.discover_tools() skill_tool_names set(skill.required_tools) return [t for t in available if t[name] in skill_tool_names or not skill_tool_names] def validate_skill(self, skill: Skill) - dict: 验证技能包的依赖是否满足。 available_tools {t[name] for t in self.mcp_client.discover_tools()} missing_tools [t for t in skill.required_tools if t not in available_tools] return { valid: not missing_tools, missing_tools: missing_tools, }MCP Skills 完整集成文件05_mcp_skills_integration.py把 MCP 和 Skills 结合起来——技能包声明需要哪些 MCP ServerSkillManager 在加载技能时自动连接对应的 Server用户消息进来 │ ▼SkillManager 选择技能包如 refund_skill │ ▼加载技能包需要的 MCP Server订单 Server 退款 Server │ ▼MCP Client 连接 Server发现可用工具 │ ▼Agent 使用这些工具处理用户问题 │ ▼返回结果释放 MCP 连接MCPSkillsAgent完整实现——接收用户消息后自动选择技能包并调用工具class MCPSkillsAgent: 结合 MCP 和 Skills 的智能客服 Agent。 def __init__(self, skill_manager: SkillManager): self.skill_manager skill_manager self.llm get_llm(temperature0) def chat(self, user_message: str) - dict: # Step 1: 选择技能包 skill self.skill_manager.select_skill(user_message) # Step 2: 获取可用工具 available_tools self.skill_manager.get_skill_tools(skill) # Step 3: 用技能包的 System Prompt 工具列表构造 Prompt tool_descriptions \n.join( f- {t[name]}: {t[description]} for t in available_tools ) system_prompt skill.system_prompt f\n\n可用工具\n{tool_descriptions} # Step 4: 多轮工具调用处理 messages [SystemMessage(contentsystem_prompt), HumanMessage(contentuser_message)] for _ in range(3): response self.llm.invoke(messages) messages.append(response) if not response.tool_calls: break for tc in response.tool_calls: result self.skill_manager.mcp_client.call_tool( tc[name], tc.get(args)) messages.append( ToolMessage(contentresult, tool_call_idtc[id])) return { reply: response.content, skill: skill.name, tools_called: [...], }这种架构的好处新增一个业务能力 新建一个 Skill 对应的 MCP Server主系统代码不用改不同环境开发/测试/生产可以连接不同的 MCP Server技能包可以独立版本管理、灰度发布代码文件文件说明01_mcp_server.py简化的 MCP Server 实现02_mcp_client.pyMCP Client发现和调用工具03_skills.pySkill 技能包定义04_skill_manager.py技能包管理器动态选择和加载05_mcp_skills_integration.pyMCP Skills 完整集成01什么是AI大模型应用开发工程师如果说AI大模型是蕴藏着巨大能量的“后台超级能力”那么AI大模型应用开发工程师就是将这种能量转化为实用工具的执行者。AI大模型应用开发工程师是基于AI大模型设计开发落地业务的应用工程师。这个职业的核心价值在于打破技术与用户之间的壁垒把普通人难以理解的算法逻辑、模型参数转化为人人都能轻松操作的产品形态。无论是日常写作时用到的AI文案生成器、修图软件里的智能美化功能还是办公场景中的自动记账工具、会议记录用的语音转文字APP这些看似简单的应用背后都是应用开发工程师在默默搭建技术与需求之间的桥梁。他们不追求创造全新的大模型而是专注于让已有的大模型“听懂”业务需求“学会”解决具体问题最终形成可落地、可使用的产品。CSDN粉丝独家福利给大家整理了一份AI大模型全套学习资料这份完整版的 AI 大模型学习资料已经上传CSDN朋友们如果需要可以扫描下方二维码点击下方CSDN官方认证链接免费领取【保证100%免费】02AI大模型应用开发工程师的核心职责需求分析与拆解是工作的起点也是确保开发不偏离方向的关键。应用开发工程师需要直接对接业务方深入理解其核心诉求——不仅要明确“要做什么”更要厘清“为什么要做”以及“做到什么程度算合格”。在此基础上他们会将模糊的业务需求拆解为具体的技术任务明确每个环节的执行标准并评估技术实现的可行性同时定义清晰的核心指标为后续开发、测试提供依据。这一步就像建筑前的图纸设计若出现偏差后续所有工作都可能白费。技术选型与适配是衔接需求与开发的核心环节。工程师需要根据业务场景的特点选择合适的基础大模型、开发框架和工具——不同的业务对模型的响应速度、精度、成本要求不同选型的合理性直接影响最终产品的表现。同时他们还要对行业相关数据进行预处理通过提示词工程优化模型输出或在必要时进行轻量化微调让基础模型更好地适配具体业务。此外设计合理的上下文管理规则确保模型理解连贯需求建立敏感信息过滤机制保障数据安全也是这一环节的重要内容。应用开发与对接则是将方案转化为产品的实操阶段。工程师会利用选定的开发框架构建应用的核心功能同时联动各类外部系统——比如将AI模型与企业现有的客户管理系统、数据存储系统打通确保数据流转顺畅。在这一过程中他们还需要配合设计团队打磨前端交互界面让技术功能以简洁易懂的方式呈现给用户实现从技术方案到产品形态的转化。测试与优化是保障产品质量的关键步骤。工程师会开展全面的功能测试找出并修复开发过程中出现的漏洞同时针对模型的响应速度、稳定性等性能指标进行优化。安全合规性也是测试的重点需要确保应用符合数据保护、隐私安全等相关规定。此外他们还会收集用户反馈通过调整模型参数、优化提示词等方式持续提升产品体验让应用更贴合用户实际使用需求。部署运维与迭代则贯穿产品的整个生命周期。工程师会通过云服务器或私有服务器将应用部署上线并实时监控运行状态及时处理突发故障确保应用稳定运行。随着业务需求的变化他们还需要对应用功能进行迭代更新同时编写完善的开发文档和使用手册为后续的维护和交接提供支持。03薪资情况与职业价值市场对这一职业的高度认可直接体现在薪资待遇上。据猎聘最新在招岗位数据显示AI大模型应用开发工程师的月薪最高可达60k。在AI技术加速落地的当下这种“技术业务”的复合型能力尤为稀缺让该职业成为当下极具吸引力的就业选择。AI大模型应用开发工程师是AI技术落地的关键桥梁。他们用专业能力将抽象的技术转化为具体的产品让大模型的价值真正渗透到各行各业。随着AI场景化应用的不断深化这一职业的重要性将更加凸显也必将吸引更多人才投身其中推动AI技术更好地服务于社会发展。CSDN粉丝独家福利给大家整理了一份AI大模型全套学习资料这份完整版的 AI 大模型学习资料已经上传CSDN朋友们如果需要可以扫描下方二维码点击下方CSDN官方认证链接免费领取【保证100%免费】