WorkBuddy MCP连接器开发实战：让你的AI助手接入企业业务系统

什么是MCP连接器？为什么企业需要它？

先抛几个核心判断。所谓MCP连接器，本质上就是为AI助手和外部系统之间搭建的一座标准化通信桥梁。Model Context Protocol（MCP）作为这套通信协议的标准，让AI助手可以安全、高效地调用企业内部系统的接口，真正实现“对话即操作”的智能化工作流。

实际业务中，这样的需求比比皆是：员工在AI助手中直接查询OA审批进度、想看看ERP系统里某个物料的库存数据、或者快速调出CRM系统里的客户信息。MCP连接器正是解决这类场景的关键——它将企业内部API封装成AI可以理解和调用的标准工具，让大模型不仅能“看到”业务数据，还能“操作”它们。

WorkBuddy MCP连接器架构概览

一个典型的MCP连接器，其架构可以拆解为三层：

协议层：负责与WorkBuddy平台建立MCP通信连接 适配层：将企业内部API转化为MCP标准工具定义（Tool Schema） 业务层：实际调用企业内部系统的HTTP/RPC接口

AI助手 ←→ WorkBuddy平台 ←→ MCP连接器 ←→ 企业内部系统(协议层)(适配层)(业务层)

开发实战：Python版OA审批查询连接器

下面用一个实际例子来演示如何开发查询OA审批状态的MCP连接器。代码中使用的均为标准占位符，实际部署时替换成企业自己的配置就好。

第一步：定义连接器配置文件

在项目根目录创建 connector.json，用于描述连接器的基本信息和工具列表：

{“name”: “oa-approval-query”,”version”: “1.0.0”,”description”: “OA审批状态查询连接器”,”transport”: “stdio”,”tools”: [{“name”: “query_approval_status”,”description”: “根据审批单号查询OA审批的当前状态，包括审批节点、审批人和审批意见”,”inputSchema”: {“type”: “object”,”properties”: {“approval_id”: {“type”: “string”,”description”: “OA审批单号”}},”required”: [“approval_id”]}}]}

第二步：实现连接器核心逻辑

#!/usr/bin/env python3”“”OA审批查询MCP连接器用于连接企业内部OA系统，提供审批状态查询能力”“”import jsonimport sysimport urllib.requestimport ssl# 企业OA系统配置（请替换为实际地址）OA_API_BASE = “https://oa.example.com/api/v1”OA_API_TOKEN = “your_api_token_here”# 请求超时配置REQUEST_TIMEOUT = 10def query_approval_status(approval_id: str) -> dict:“”“调用企业OA系统API查询审批状态”“”url = f”{OA_API_BASE}/approvals/{approval_id}”headers = {“Content-Type”: “application/json”,”Authorization”: f”Bearer {OA_API_TOKEN}”}# 创建SSL上下文（生产环境请配置有效证书）ctx = ssl.create_default_context()ctx.check_hostname = Falsectx.verify_mode = ssl.CERT_NONEreq = urllib.request.Request(url, headers=headers, method=”GET”)try:with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT, context=ctx) as resp:data = json.loads(resp.read().decode(“utf-8”))return {“success”: True,”approval_id”: data.get(“id”),”status”: data.get(“status”),”current_node”: data.get(“current_node”),”applicant”: data.get(“applicant”),”history”: data.get(“approval_history”, [])}except urllib.error.HTTPError as e:return {“success”: False,”error”: f”OA接口返回错误: {e.code} {e.reason}”}except Exception as e:return {“success”: False,”error”: f”请求异常: {str(e)}”}def handle_tool_call(tool_name: str, arguments: dict) -> str:“”“MCP工具调用路由”“”if tool_name == “query_approval_status”:result = query_approval_status(arguments[“approval_id”])return json.dumps(result, ensure_ascii=False)else:return json.dumps({“error”: f”未知工具: {tool_name}”})def main():“”“MCP连接器主循环：通过stdio与WorkBuddy平台通信”“”for line in sys.stdin:line = line.strip()if not line:continuetry:message = json.loads(line)if message.get(“method”) == “tools/call”:tool_name = message[“params”][“name”]arguments = message[“params”].get(“arguments”, {})result = handle_tool_call(tool_name, arguments)response = {“jsonrpc”: “2.0”,”id”: message.get(“id”),”result”: {“content”: [{“type”: “text”, “text”: result}]}}print(json.dumps(response), flush=True)except json.JSONDecodeError:print(json.dumps({“jsonrpc”: “2.0”,”id”: None,”error”: {“code”: -32700, “message”: “JSON解析错误”}}), flush=True)if __name__ == “__main__”:main()

第三步：在WorkBuddy中注册连接器

将连接器部署到服务器后，在WorkBuddy平台的管理后台进行注册。注册时需要配置连接器的启动命令、工作目录以及允许访问的工具列表。配置完成后，AI助手即可在对话中自动调用审批查询能力。

开发实战：Node.js版ERP库存查询连接器

对于使用Node.js技术栈的团队，开发MCP连接器的流程同样简洁：

#!/usr/bin/env node/** * ERP库存查询MCP连接器 * 连接企业ERP系统，提供实时库存数据查询 */const https = require(“https”);// 企业ERP系统配置（请替换为实际地址）const ERP_API_BASE = “https://erp.example.com/api/v2”;const ERP_API_KEY = “your_api_key_here”;function httpsGet(url, headers) {return new Promise((resolve, reject) => {const req = https.get(url, { headers }, (res) => {let data = “”;res.on(“data”, (chunk) => (data += chunk));res.on(“end”, () => resolve(JSON.parse(data)));});req.on(“error”, reject);req.setTimeout(10000, () => { req.destroy(); reject(new Error(“请求超时”)); });});}async function queryInventory(skuId) {const url = `{ERP_API_BASE}/inventory/{skuId}`;const headers = {“X-API-Key”: ERP_API_KEY};return await httpsGet(url, headers);}// MCP消息处理循环process.stdin.setEncoding(“utf8”);let buffer = “”;process.stdin.on(“data”, (chunk) => {buffer += chunk;const lines = buffer.split(” ”);buffer = lines.pop();for (const line of lines) {if (!line.trim()) continue;try {const msg = JSON.parse(line);if (msg.method === “tools/call”) {const { name, arguments: args } = msg.params;queryInventory(args.sku_id).then((result) => {const resp = {jsonrpc: “2.0”,id: msg.id,result: {content: [{ type: “text”, text: JSON.stringify(result) }]}};process.stdout.write(JSON.stringify(resp) + “ ”);}).catch((err) => {const resp = {jsonrpc: “2.0”,id: msg.id,result: {content: [{ type: “text”, text: JSON.stringify({ error: err.message }) }]}};process.stdout.write(JSON.stringify(resp) + “ ”);});}} catch (e) {// 忽略非法JSON行}}});

开发注意事项

安全方面，MCP连接器作为AI助手访问企业系统的通道，安全管控始终是首要考量。建议在生产环境中启用TLS加密通信，API密钥通过环境变量注入而非硬编码，并对每个工具设置独立的调用权限。

错误处理这块也不能忽视。企业内部API可能出现超时、服务不可用等情况，连接器需要做好降级处理，返回明确的错误信息供AI助手理解并反馈给用户。

可观测性方面，建议为连接器添加日志记录，记录每次工具调用的参数、返回结果和耗时，便于后续排查和优化。

扩展方向

完成基础连接器开发后，可以考虑以下扩展能力：

批量查询：支持一次性查询多个审批单或SKU的聚合工具 写入操作：在严格的权限管控下，支持AI助手发起审批、修改库存等写入操作 流式响应：对于大数据量查询场景，支持流式返回结果