企业协作

企业协作平台接入详细教程

作者 winter.yu
2026年3月7日 6 分钟阅读
0

企业协作平台接入详细教程

目录

  1. 概述
  2. 钉钉接入
  3. 企业微信接入
  4. 飞书接入
  5. 通用 webhook 方案
  6. 最佳实践

概述

企业协作平台通常指需要集成到企业IM工具（钉钉、企业微信、飞书）的AI协作助手，常见场景包括：

  • 智能协作：根据提示生成文案、邮件、报告
  • 文档润色：优化已有文本的语法、风格
  • 内容续写：基于前文自动补全
  • 多轮对话：支持上下文理解的协作辅助

钉钉接入

方式一：钉钉群机器人（最简单，推荐入门）

1. 创建机器人

  1. 打开钉钉群 → 群设置 → 智能群助手
  2. 点击「添加机器人」→「自定义（通过Webhook接入自定义服务）」
  3. 设置机器人名称、头像，选择安全设置（建议加签或IP白名单）
  4. 复制 Webhook 地址 和 加签密钥（如有设置）

2. 加签算法（Python示例）

import time
import hmac
import hashlib
import base64
import urllib.parse

def generate_sign(secret):
    """钉钉机器人加签算法"""
    timestamp = str(round(time.time() * 1000))
    string_to_sign = f'{timestamp}\n{secret}'
    hmac_code = hmac.new(
        secret.encode('utf-8'),
        string_to_sign.encode('utf-8'),
        digestmod=hashlib.sha256
    ).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return timestamp, sign

# 使用示例
secret = 'SECxxxxxxxxxxxxxxxxxxxxx'
timestamp, sign = generate_sign(secret)
print(f"timestamp={timestamp}&sign={sign}")

3. 发送消息

import requests
import json

webhook = 'https://oapi.dingtalk.com/robot/send?access_token=xxxxx'
secret = 'SECxxxxxxxxxx'

timestamp, sign = generate_sign(secret)
url = f"{webhook}&timestamp={timestamp}&sign={sign}"

# 发送文本消息
payload = {
    "msgtype": "text",
    "text": {
        "content": "你好，我是AI协作助手！"
    },
    "at": {
        "isAtAll": False
    }
}

# 发送Markdown
payload_md = {
    "msgtype": "markdown",
    "markdown": {
        "title": "协作结果",
        "text": "## 生成的文案\n\n这是AI生成的内容..."
    }
}

response = requests.post(url, json=payload)
print(response.json())

4. 接收用户消息（回调）

机器人需要部署一个 HTTP 服务接收钉钉推送的消息：

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/dingtalk/callback', methods=['POST'])
def dingtalk_callback():
    data = request.get_json()

    # 处理不同的事件类型
    msg_type = data.get('msgtype')

    if msg_type == 'text':
        user_content = data.get('text', {}).get('content')
        sender_staff_id = data.get('senderStaffId')

        # 调用你的协作AI处理
        ai_response = process_writing_request(user_content)

        # 回复消息
        send_reply(data.get('sessionWebhook'), ai_response)

    return jsonify({"status": "ok"})

def process_writing_request(content):
    """调用你的协作AI逻辑"""
    # 这里接入你的AI服务
    return f"收到请求：{content}\n\nAI生成结果：..."

def send_reply(webhook, message):
    """通过sessionWebhook回复"""
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": "AI协作助手",
            "text": message
        }
    }
    requests.post(webhook, json=payload)

if __name__ == '__main__':
    app.run(port=8080)

方式二：钉钉企业内部应用（功能最全）

适合需要获取组织架构、发送工作通知等高级场景。

1. 创建应用

  1. 登录 钉钉开放平台
  2. 创建「企业内部应用」
  3. 获取 AppKey 和 AppSecret
  4. 配置权限：如「通讯录管理」、「消息通知」等

2. 获取 AccessToken

import requests

APP_KEY = 'dingxxxxxxxxxx'
APP_SECRET = 'xxxxxxxxxxxxxx'

def get_access_token():
    url = f"https://oapi.dingtalk.com/gettoken?appkey={APP_KEY}&appsecret={APP_SECRET}"
    response = requests.get(url)
    return response.json().get('access_token')

token = get_access_token()

3. 发送工作通知（异步消息）

def send_work_notification(token, user_id_list, message):
    """发送工作通知到指定用户"""
    url = f"https://oapi.dingtalk.com/topapi/message/corpconversation/asyncsend_v2?access_token={token}"

    payload = {
        "agent_id": "你的AgentID",
        "userid_list": user_id_list,  # 逗号分隔的用户ID
        "msg": {
            "msgtype": "markdown",
            "markdown": {
                "title": "AI协作助手",
                "text": message
            }
        }
    }

    response = requests.post(url, json=payload)
    return response.json()

企业微信接入

方式一：群机器人

1. 创建机器人

  1. 打开企业微信群 → 右上角「…」→「添加群机器人」
  2. 设置机器人名称和头像
  3. 复制 Webhook 地址

2. 发送消息

import requests

WEBHOOK_URL = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxxx'

def send_text_message(content, mentioned_list=None):
    """发送文本消息"""
    payload = {
        "msgtype": "text",
        "text": {
            "content": content,
            "mentioned_list": mentioned_list or []
        }
    }
    response = requests.post(WEBHOOK_URL, json=payload)
    return response.json()

def send_markdown(content):
    """发送Markdown消息"""
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "content": content
        }
    }
    response = requests.post(WEBHOOK_URL, json=payload)
    return response.json()

def send_news_card(title, description, url, picurl=None):
    """发送图文卡片"""
    payload = {
        "msgtype": "news",
        "news": {
            "articles": [
                {
                    "title": title,
                    "description": description,
                    "url": url,
                    "picurl": picurl or ""
                }
            ]
        }
    }
    response = requests.post(WEBHOOK_URL, json=payload)
    return response.json()

# 使用示例
send_text_message("AI协作助手已就绪！")
send_markdown("**协作结果**\n> 生成的文案内容...")

3. 上传文件/图片

def upload_media(media_type, file_path):
    """上传临时素材，返回media_id"""
    url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/upload_media?key=YOUR_KEY&type={media_type}"

    with open(file_path, 'rb') as f:
        files = {'media': f}
        response = requests.post(url, files=files)

    return response.json().get('media_id')

def send_file(media_id):
    """发送文件"""
    payload = {
        "msgtype": "file",
        "file": {
            "media_id": media_id
        }
    }
    response = requests.post(WEBHOOK_URL, json=payload)
    return response.json()

方式二：自建应用（接收用户消息）

1. 创建应用

  1. 登录 企业微信管理后台
  2. 应用管理 → 创建应用
  3. 获取 AgentId 和 Secret
  4. 设置接收消息：
    • URL：你的服务器地址
    • Token：随机字符串，用于验证
    • EncodingAESKey：消息加密密钥

2. 验证URL（服务器配置）

from flask import Flask, request
import hashlib
import xml.etree.ElementTree as ET
import base64
from Crypto.Cipher import AES
import random
import string

app = Flask(__name__)

TOKEN = 'your_token'
ENCODING_AES_KEY = 'your_encoding_aes_key'
CORP_ID = 'your_corp_id'

def verify_url(msg_signature, timestamp, nonce, echostr):
    """验证URL并解密echostr"""
    # 排序拼接
    tmp_list = [TOKEN, timestamp, nonce, echostr]
    tmp_list.sort()
    tmp_str = ''.join(tmp_list)

    # SHA1签名
    signature = hashlib.sha1(tmp_str.encode()).hexdigest()

    if signature != msg_signature:
        return None

    # 解密echostr
    aes_key = base64.b64decode(ENCODING_AES_KEY + '=')
    cipher = AES.new(aes_key, AES.MODE_CBC, aes_key[:16])
    decrypted = cipher.decrypt(base64.b64decode(echostr))

    # 去掉填充
    pad_len = decrypted[-1]
    content = decrypted[16:-pad_len]

    # 去掉CorpID
    if content[-len(CORP_ID):] == CORP_ID.encode():
        content = content[:-len(CORP_ID)]

    return content

@app.route('/wechat/callback', methods=['GET', 'POST'])
def wechat_callback():
    if request.method == 'GET':
        # 验证URL
        signature = request.args.get('msg_signature')
        timestamp = request.args.get('timestamp')
        nonce = request.args.get('nonce')
        echostr = request.args.get('echostr')

        result = verify_url(signature, timestamp, nonce, echostr)
        return result if result else 'fail'

    else:
        # 处理接收到的消息
        # TODO: 解密XML消息，调用AI处理，回复结果
        pass

3. 发送应用消息

import requests

CORP_ID = 'wwxxxxxxxxxxxxxxxx'
SECRET = 'xxxxxxxxxxxxxxxxxx'
AGENT_ID = 1000002

def get_access_token():
    url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={CORP_ID}&corpsecret={SECRET}"
    response = requests.get(url)
    return response.json().get('access_token')

def send_message(token, user_id, message):
    """发送应用消息"""
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"

    payload = {
        "touser": user_id,
        "msgtype": "text",
        "agentid": AGENT_ID,
        "text": {
            "content": message
        },
        "safe": 0
    }

    response = requests.post(url, json=payload)
    return response.json()

# Markdown消息
payload_md = {
    "touser": user_id,
    "msgtype": "markdown",
    "agentid": AGENT_ID,
    "markdown": {
        "content": "**AI协作助手**\n> 生成内容：..."
    }
}

飞书接入

飞书（Lark）API 设计较为现代化，推荐使用。

方式一：群机器人

1. 创建机器人

  1. 打开飞书群 → 设置 → 群机器人 → 添加机器人
  2. 选择「自定义机器人」
  3. 复制 Webhook 地址 和 签名校验密钥

2. 发送消息

import requests
import json

WEBHOOK_URL = 'https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxx'

def send_text(content):
    """发送文本消息"""
    payload = {
        "msg_type": "text",
        "content": {
            "text": content
        }
    }
    response = requests.post(WEBHOOK_URL, json=payload)
    return response.json()

def send_rich_text(title, content):
    """发送富文本消息"""
    payload = {
        "msg_type": "post",
        "content": {
            "post": {
                "zh_cn": {
                    "title": title,
                    "content": [
                        [{
                            "tag": "text",
                            "text": content
                        }]
                    ]
                }
            }
        }
    }
    response = requests.post(WEBHOOK_URL, json=payload)
    return response.json()

def send_card(title, content):
    """发送卡片消息"""
    payload = {
        "msg_type": "interactive",
        "card": {
            "config": {"wide_screen_mode": True},
            "header": {
                "title": {
                    "tag": "plain_text",
                    "content": title
                }
            },
            "elements": [
                {
                    "tag": "div",
                    "text": {
                        "tag": "lark_md",
                        "content": content
                    }
                }
            ]
        }
    }
    response = requests.post(WEBHOOK_URL, json=payload)
    return response.json()

方式二：自建应用（推荐生产环境）

1. 创建应用

  1. 登录 飞书开放平台
  2. 创建企业自建应用
  3. 获取 App ID 和 App Secret
  4. 订阅「接收消息」事件

2. 获取 Tenant Access Token

def get_tenant_access_token(app_id, app_secret):
    """获取租户级Token（企业内使用）"""
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = {
        "app_id": app_id,
        "app_secret": app_secret
    }
    response = requests.post(url, json=payload)
    return response.json().get('tenant_access_token')

APP_ID = 'cli_xxxxxx'
APP_SECRET = 'xxxxxxxx'
token = get_tenant_access_token(APP_ID, APP_SECRET)

3. 接收消息事件

from flask import Flask, request, jsonify
import hashlib
import json

app = Flask(__name__)
VERIFICATION_TOKEN = 'your_verification_token'
ENCRYPT_KEY = 'your_encrypt_key'

@app.route('/feishu/callback', methods=['POST'])
def feishu_callback():
    data = request.get_json()

    # 验证请求
    token = data.get('token')
    if token != VERIFICATION_TOKEN:
        return jsonify({"code": 403})

    # 处理URL验证
    if data.get('type') == 'url_verification':
        return jsonify({"challenge": data.get('challenge')})

    # 处理事件回调
    event = data.get('event', {})
    event_type = event.get('type')

    if event_type == 'im.message.receive_v1':
        message = event.get('message', {})
        content = json.loads(message.get('content', '{}'))
        text = content.get('text', '')
        chat_id = message.get('chat_id')

        # 调用AI处理
        ai_response = process_writing(text)

        # 回复消息
        send_message_reply(token, chat_id, ai_response)

    return jsonify({"code": 0})

def send_message_reply(token, chat_id, content):
    """回复消息"""
    url = "https://open.feishu.cn/open-apis/message/v4/send"
    headers = {"Authorization": f"Bearer {token}"}
    payload = {
        "chat_id": chat_id,
        "msg_type": "text",
        "content": {
            "text": content
        }
    }
    requests.post(url, headers=headers, json=payload)

通用 Webhook 方案

如果不想深入各平台的SDK，可以使用统一的 Webhook 转发服务：

# webhook_gateway.py - 统一的Webhook网关

from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

@app.route('/gateway/
<platform>', methods=['POST'])
def gateway(platform):
    """统一的Webhook入口"""
    data = request.get_json()
    user_message = data.get('message')

    # 调用你的协作AI
    ai_result = call_writing_ai(user_message)

    # 根据平台路由
    if platform == 'dingtalk':
        return dingtalk_reply(data, ai_result)
    elif platform == 'wechat':
        return wechat_reply(data, ai_result)
    elif platform == 'feishu':
        return feishu_reply(data, ai_result)

    return jsonify({"error": "unknown platform"}), 400

def call_writing_ai(prompt):
    """调用你的协作AI服务"""
    # 接入OpenAI/Claude/国产大模型等
    # 返回AI生成结果
    return "这是AI生成的内容..."

最佳实践

1. 消息长度限制

平台 文本消息限制 Markdown限制
钉钉 4096字符 同文本
企业微信 2048字节 4096字节
飞书 无明确限制 无明确限制

建议：超过限制时拆分消息或使用文件发送。

2. 安全建议

# 使用环境变量存储敏感信息
import os

DINGTALK_SECRET = os.getenv('DINGTALK_SECRET')
WECHAT_KEY = os.getenv('WECHAT_KEY')
FEISHU_TOKEN = os.getenv('FEISHU_TOKEN')

3. 消息模板设计

WRITING_RESULT_TEMPLATE = """
📝 **协作结果**

{content}

---
💡 提示：回复「重写」可重新生成
"""

ERROR_TEMPLATE = """
❌ **处理失败**

{error_message}

请稍后重试或联系管理员
"""

4. 速率限制

平台 频率限制
钉钉 20次/秒
企业微信 20次/秒
飞书 100次/秒

5. 完整的快速启动示例

# writing_bot.py - 一个极简的协作机器人

import os
import requests
from flask import Flask, request

app = Flask(__name__)

# 配置
DINGTALK_WEBHOOK = os.getenv('DINGTALK_WEBHOOK')
DINGTALK_SECRET = os.getenv('DINGTALK_SECRET')

# 模拟AI协作（替换为你的真实AI调用）
def ai_write(prompt):
    return f"🤖 根据「{prompt}」生成的内容：\n\n这里是AI生成的文案..."

@app.route('/write', methods=['POST'])
def write():
    """接收协作请求"""
    data = request.get_json()
    prompt = data.get('prompt', '')
    webhook = data.get('webhook', DINGTALK_WEBHOOK)

    # 生成内容
    result = ai_write(prompt)

    # 发送结果
    send_to_dingtalk(webhook, result)

    return {"status": "ok"}

def send_to_dingtalk(webhook, message):
    """发送到钉钉"""
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": "AI协作结果",
            "text": message
        }
    }
    requests.post(webhook, json=payload)

if __name__ == '__main__':
    app.run(port=5000)

参考文档

💡 提示：根据你的实际场景选择接入方式：

  • 快速验证/小团队 → 群机器人
  • 生产环境/需要用户交互 → 自建应用
  • 多平台统一 → Webhook网关方案

标签：

作者

winter.yu

关注我
其他文章
暂无评论！成为第一个。

发表回复