本章目录

  1. MCP 是什么,为什么重要
  2. 实现一个简单的 MCP Server
  3. MCP Client 调用工具
  4. Skills 技能包设计
  5. 技能包动态加载
  6. MCP + Skills 完整集成

  1. 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 自动就能用

  1. 实现一个简单的 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"}'}]}

  1. 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 代码不用改

  1. 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 默认加载自己的专属技能包,需要时也可以借用其他技能包

  1. 技能包动态加载

文件: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,        }

  1. 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%免费】

在这里插入图片描述

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐