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

MCP（Model Context Protocol）是AI助手与外部系统之间的标准化通信协议。通过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 json
import sys
import urllib.request
import ssl

# 企业OA系统配置（请替换为实际地址）
OA_API_BASE = "https://oa.example.com/api/v1"
OA_API_TOKEN = "your_api_token_here"

# 请求超时配置
REQUEST_TIMEOUT = 10


def 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 = False
    ctx.verify_mode = ssl.CERT_NONE

    req = 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:
            continue

        try:
            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("\n");
  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) + "\n");
          })
          .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) + "\n");
          });
      }
    } catch (e) {
      // 忽略非法JSON行
    }
  }
});

开发注意事项

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

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

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

扩展方向

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

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

更多MCP连接器开发规范和高级用法，可参考WorkBuddy官方文档。

关于华万通信

上海华万通信科技有限公司，专注为企业提供腾讯系SaaS产品的一站式选型与集成服务，包括腾讯会议、企业微信、腾讯电子签等。公司致力于帮助企业客户实现数字化办公的智能化升级，提供从产品选型、技术集成到持续运营的全流程服务支持。