AI客服系统如何实现灵活扩展?MCP协议+Skills体系深度解析
本文介绍了MCP(Model Context Protocol)协议及其实现方法。MCP作为AI领域的"USB接口",解决了工具接入需要重复适配的问题。主要内容包括:1)MCP的核心价值是标准化AI应用与工具间的交互;2)通过Python代码演示了MCP Server的实现,包括工具注册和请求处理;3)展示了MCP Client如何连接多个Server并统一调用工具。该协议使业务团队能独立开发工具
本章目录
- MCP 是什么,为什么重要
- 实现一个简单的 MCP Server
- MCP Client 调用工具
- Skills 技能包设计
- 技能包动态加载
- MCP + Skills 完整集成
- MCP 是什么,为什么重要
MCP(Model 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 Server,Agent 自动就能用
- 实现一个简单的 MCP Server
文件:01_mcp_server.py
MCP 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_ascii=False) try: result = tool["handler"](**arguments) if isinstance(result, dict): return json.dumps(result, ensure_ascii=False) return str(result) except TypeError as e: return json.dumps({"error": f"参数错误: {e}"}, ensure_ascii=False) except Exception as e: return json.dumps({"error": f"执行错误: {e}"}, ensure_ascii=False) 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 Server:
order_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.py
MCP 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_ascii=False) 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( name="refund", 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( name="logistics", description="查询物流和订单配送状态", system_prompt="你是星辰科技的物流查询专员。处理流程:1. 查订单 2. 查物流 3. 告知详情。", required_tools=["query_order", "query_logistics"], required_servers=["order_service"],)# 技术支持技能包tech_skill = Skill( name="tech_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.py
SkillManager 负责:根据用户问题,自动选择最合适的技能包加载:
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(temperature=0) skill_list = "\n".join( f"- {name}: {skill.description}" for name, skill in self.skills.items() ) prompt = f"""根据用户消息,选择最合适的技能包。可用技能包:{skill_list}请只输出技能包名称,不要输出其他内容。""" response = llm.invoke([HumanMessage(content=prompt)]) 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 Server,SkillManager 在加载技能时自动连接对应的 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(temperature=0) 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(content=system_prompt), HumanMessage(content=user_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(content=result, tool_call_id=tc["id"])) return { "reply": response.content, "skill": skill.name, "tools_called": [...], }
这种架构的好处:
- 新增一个业务能力 = 新建一个 Skill + 对应的 MCP Server,主系统代码不用改
- 不同环境(开发/测试/生产)可以连接不同的 MCP Server
- 技能包可以独立版本管理、灰度发布
代码文件
| 文件 | 说明 |
|---|---|
01_mcp_server.py |
简化的 MCP Server 实现 |
02_mcp_client.py |
MCP Client,发现和调用工具 |
03_skills.py |
Skill 技能包定义 |
04_skill_manager.py |
技能包管理器,动态选择和加载 |
05_mcp_skills_integration.py |
MCP + Skills 完整集成 |
01
什么是AI大模型应用开发工程师?
如果说AI大模型是蕴藏着巨大能量的“后台超级能力”,那么AI大模型应用开发工程师就是将这种能量转化为实用工具的执行者。
AI大模型应用开发工程师是基于AI大模型,设计开发落地业务的应用工程师。
这个职业的核心价值,在于打破技术与用户之间的壁垒,把普通人难以理解的算法逻辑、模型参数,转化为人人都能轻松操作的产品形态。
无论是日常写作时用到的AI文案生成器、修图软件里的智能美化功能,还是办公场景中的自动记账工具、会议记录用的语音转文字APP,这些看似简单的应用背后,都是应用开发工程师在默默搭建技术与需求之间的桥梁。
他们不追求创造全新的大模型,而是专注于让已有的大模型“听懂”业务需求,“学会”解决具体问题,最终形成可落地、可使用的产品。
CSDN粉丝独家福利
给大家整理了一份AI大模型全套学习资料,这份完整版的 AI 大模型学习资料已经上传CSDN,朋友们如果需要可以扫描下方二维码&点击下方CSDN官方认证链接免费领取 【保证100%免费】

02
AI大模型应用开发工程师的核心职责
需求分析与拆解是工作的起点,也是确保开发不偏离方向的关键。
应用开发工程师需要直接对接业务方,深入理解其核心诉求——不仅要明确“要做什么”,更要厘清“为什么要做”以及“做到什么程度算合格”。
在此基础上,他们会将模糊的业务需求拆解为具体的技术任务,明确每个环节的执行标准,并评估技术实现的可行性,同时定义清晰的核心指标,为后续开发、测试提供依据。
这一步就像建筑前的图纸设计,若出现偏差,后续所有工作都可能白费。
技术选型与适配是衔接需求与开发的核心环节。
工程师需要根据业务场景的特点,选择合适的基础大模型、开发框架和工具——不同的业务对模型的响应速度、精度、成本要求不同,选型的合理性直接影响最终产品的表现。
同时,他们还要对行业相关数据进行预处理,通过提示词工程优化模型输出,或在必要时进行轻量化微调,让基础模型更好地适配具体业务。
此外,设计合理的上下文管理规则确保模型理解连贯需求,建立敏感信息过滤机制保障数据安全,也是这一环节的重要内容。
应用开发与对接则是将方案转化为产品的实操阶段。
工程师会利用选定的开发框架构建应用的核心功能,同时联动各类外部系统——比如将AI模型与企业现有的客户管理系统、数据存储系统打通,确保数据流转顺畅。
在这一过程中,他们还需要配合设计团队打磨前端交互界面,让技术功能以简洁易懂的方式呈现给用户,实现从技术方案到产品形态的转化。
测试与优化是保障产品质量的关键步骤。
工程师会开展全面的功能测试,找出并修复开发过程中出现的漏洞,同时针对模型的响应速度、稳定性等性能指标进行优化。
安全合规性也是测试的重点,需要确保应用符合数据保护、隐私安全等相关规定。
此外,他们还会收集用户反馈,通过调整模型参数、优化提示词等方式持续提升产品体验,让应用更贴合用户实际使用需求。
部署运维与迭代则贯穿产品的整个生命周期。
工程师会通过云服务器或私有服务器将应用部署上线,并实时监控运行状态,及时处理突发故障,确保应用稳定运行。
随着业务需求的变化,他们还需要对应用功能进行迭代更新,同时编写完善的开发文档和使用手册,为后续的维护和交接提供支持。
03
薪资情况与职业价值
市场对这一职业的高度认可,直接体现在薪资待遇上。
据猎聘最新在招岗位数据显示,AI大模型应用开发工程师的月薪最高可达60k。

在AI技术加速落地的当下,这种“技术+业务”的复合型能力尤为稀缺,让该职业成为当下极具吸引力的就业选择。
AI大模型应用开发工程师是AI技术落地的关键桥梁。
他们用专业能力将抽象的技术转化为具体的产品,让大模型的价值真正渗透到各行各业。
随着AI场景化应用的不断深化,这一职业的重要性将更加凸显,也必将吸引更多人才投身其中,推动AI技术更好地服务于社会发展。
CSDN粉丝独家福利
给大家整理了一份AI大模型全套学习资料,这份完整版的 AI 大模型学习资料已经上传CSDN,朋友们如果需要可以扫描下方二维码&点击下方CSDN官方认证链接免费领取 【保证100%免费】

更多推荐

所有评论(0)