最近做的企微二开,要接入 OpenClaw 作为 AI 引擎——OpenClaw 负责对话理解和工具调用,企微 API 负责消息收发和客户管理,两边各干各的、通过中间层协同。这篇重点讲 OpenClaw 和企微接口怎么对接——消息怎么从企微传到 OpenClaw、OpenClaw 的工具调用怎么映射到企微接口、对话状态怎么同步。把调了哪些接口记下来。

Eyun 平台开放的企微 API,统一 POST+JSON,鉴权用 App Token 加 appid(Authorization: Bearer eyk_xxxx),路径 {BASE_URL}/wx-api/api/<模块>/<动作>,响应封套 {code, data, detail, message, time},code 为 0 成功。

消息流转:企微到 OpenClaw

客户在企微发消息,先通过 Webhook 回调到我们后端,后端转发给 OpenClaw 处理,OpenClaw 生成回复后我们再调企微接口发回去:

from flask import Flask, request
import requests

app = Flask(__name__)
BASE = "https://api.eyun.com"
HEADERS = {"Authorization": "Bearer eyk_xxxx", "Content-Type": "application/json"}

@app.route("/wx-api/webhook/", methods=["POST"])
def webhook():
    event = request.headers.get("X-Eyun-Event")
    if event != "message":
        return "ok"
    payload = request.json["data"]
    appid = payload["appid"]
    from_uin = payload["fromUin"]
    content = payload["content"]
    
    # 1. 转发给OpenClaw处理
    reply = call_openclaw(appid, from_uin, content)
    
    # 2. 调企微接口发回复
    requests.post(
        f"{BASE}/wx-api/api/message/sendText",
        headers=HEADERS,
        json={"appid": appid, "to": from_uin, "content": reply}
    )
    return "ok"

OpenClaw 不直接调企微接口,所有企微接口调用都在我们后端。OpenClaw 只负责理解和生成,我们后端负责收发。这样 OpenClaw 不需要知道企微的鉴权和接口细节,耦合低。回调机制在Eyun 开发文档。

OpenClaw 工具调用映射到企微接口

OpenClaw 输出工具调用请求,我们后端把工具调用映射到企微接口。工具映射表:

def call_openclaw(appid, from_uin, content):
    # 传给OpenClaw的上下文
    context = {
        "customer_uin": from_uin,
        "appid": appid,
        "history": get_history(from_uin)
    }
    
    # OpenClaw返回工具调用或直接回复
    result = openclaw.chat(content, context, tools=TOOL_DEFS)
    
    if result["type"] == "tool_call":
        # 执行工具(调企微接口)
        tool_result = execute_tool(appid, result["tool_name"], result["args"])
        # 把工具结果给OpenClaw继续生成
        return openclaw.continue_with_tool_result(tool_result)
    else:
        return result["content"]

def execute_tool(appid, tool_name, args):
    if tool_name == "query_customer":
        return query_customer(appid, args["keyword"])
    elif tool_name == "send_message":
        return send_message(appid, args["to_uin"], args["content"])

工具映射让 OpenClaw 不直接碰企微接口,我们控制工具的执行——可以加权限校验、日志记录、频率控制。

客户查询工具:调联系人接口

OpenClaw 要查客户信息时,工具执行体调企微联系人接口:

def query_customer(appid, keyword):
    if keyword.isdigit() and len(keyword) >= 11:
        resp = requests.post(
            f"{BASE}/wx-api/api/contact/phoneNumberSearch",
            headers=HEADERS,
            json={"appid": appid, "phone": keyword}
        )
    else:
        resp = requests.post(
            f"{BASE}/wx-api/api/contact/search",
            headers=HEADERS,
            json={"appid": appid, "keyword": keyword}
        )
    data = resp.json()["data"]
    return {"name": data.get("nickName", ""), "phone": data.get("mobile", "")}

返回结构化数据给 OpenClaw,OpenClaw 基于数据生成自然语言回复。。

对话状态同步

OpenClaw 维护对话状态,但状态 key 要和企微的 uin 对应。每次调 OpenClaw 带上 from_uin 作为 session key:

def call_openclaw(appid, from_uin, content):
    session_id = f"wecom:{appid}:{from_uin}"
    return openclaw.chat(content, session_id=session_id)

session_id 用 appid 加 uin 保证唯一——同一客户在不同 appid 下的对话不串。OpenClaw 内部维护对话历史和上下文,我们不用自己管。

工具调用的权限控制

OpenClaw 想调什么工具就调什么会出事。工具执行前要权限校验:

def execute_tool_safe(appid, tool_name, args, caller_uin):
    # 查调用者权限
    staff = get_staff_profile(appid, caller_uin)
    
    if tool_name == "send_message" and staff["role"] != "sales":
        return {"error": "无权限发消息"}
    if tool_name == "update_label" and staff["role"] == "guest":
        return {"error": "无权限打标签"}
    
    return execute_tool(appid, tool_name, args)

权限校验在工具执行层做,不依赖 OpenClaw 自觉。OpenClaw 可能生成不合规的工具调用,我们拦住就行。

写在最后

接入 OpenClaw 这套东西,本质是做个中间层——企微 Webhook 收消息转发给 OpenClaw、OpenClaw 输出工具调用映射到企微接口执行、对话状态用 uin 做 key 同步。OpenClaw 不直接碰企微接口,所有接口调用在我们后端控制,能加权限、日志、限流。接口路径、参数、回调机制在开发文档里。凭证和接入地址在Eyun 企业微信 API 平台开通。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐