Claude API 接入飞书的技术实现与避坑指南

1次阅读

没有评论

共计 2844 个字符，预计需要花费 8 分钟才能阅读完成。

企业 IM 集成 AI 助手正成为提升工作效率的热门需求。飞书作为企业级协作平台，通过开放 API 支持第三方服务集成。而 Claude 作为强大的对话 AI，能够处理复杂问答和文本分析。将两者结合，可以打造智能化的办公助手。

但在实际开发中，我们遇到了几个主要挑战：

认证流程复杂：需要同时处理飞书 OAuth2.0 和 Claude API Key 双重认证
消息格式差异：飞书的消息结构与 Claude 的输入输出格式需要转换
异步处理需求：AI 响应时间可能较长，需要保持飞书会话不超时
性能瓶颈：高并发场景下的请求处理和限流策略
安全性要求：企业数据需要端到端保护

在接入方案上，我们对比了两种主流方式：

Webhook 方案
优点：部署简单，适合轻量级应用
缺点：需要处理签名验证，调试困难
SDK 方案
优点：官方维护，功能完善
缺点：依赖较重，升级成本高

综合考虑后，我们选择了混合方案：

使用飞书官方 SDK 处理基础认证
自定义 Webhook 处理业务逻辑
自建中间层做协议转换

登录飞书开放平台，创建自建应用
配置权限：im:message、im:chat 等
设置事件订阅：接收用户消息
配置安全域名和 IP 白名单

获取 API Key 并设置环境变量
实现请求签名生成方法
处理 Token 自动刷新逻辑

# 示例：Claude 认证封装
class ClaudeClient:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.anthropic.com/v1"

    def _generate_headers(self):
        return {
            "x-api-key": self.api_key,
            "anthropic-version": "2023-06-01",
            "Content-Type": "application/json"
        }

飞书消息到 Claude 输入的转换逻辑：

提取飞书消息中的文本内容
处理 @提及和富文本格式
构建 Claude 要求的 prompt 结构

def convert_feishu_to_claude(feishu_msg):
    """转换飞书消息为 Claude 输入格式"""
    content = feishu_msg["event"]["message"]["content"]
    # 处理富文本 JSON
    if isinstance(content, str):
        try:
            content = json.loads(content)
        except:
            pass

    # 提取纯文本
    text = ""
    if isinstance(content, dict):
        text = content.get("text", "")
    else:
        text = str(content)

    return {"prompt": f"\n\nHuman: {text}\n\nAssistant:",
        "max_tokens_to_sample": 1000
    }

采用 Celery+Redis 实现异步任务队列：

飞书 Webhook 接收消息后立即返回
创建异步任务处理 AI 请求
通过飞书消息 API 异步回复

@app.route('/feishu/webhook', methods=['POST'])
def feishu_webhook():
    # 验证签名
    if not verify_signature(request):
        return jsonify({"error": "Invalid signature"}), 403

    # 处理事件订阅验证
    if request.json.get("type") == "url_verification":
        return jsonify({"challenge": request.json.get("challenge")})

    # 放入消息队列
    task = process_message.delay(request.json)

    return jsonify({"msg": "processing"})

class ClaudeClient:
    # ... 初始化代码见上文...

    def complete(self, prompt, **kwargs):
        """调用 Claude 完成文本生成"""
        data = {"prompt": prompt, **kwargs}

        try:
            resp = requests.post(f"{self.base_url}/complete",
                headers=self._generate_headers(),
                json=data,
                timeout=30
            )
            resp.raise_for_status()
            return resp.json()["completion"]
        except Exception as e:
            logger.error(f"Claude API error: {str(e)}")
            raise

@celery.task(bind=True)
def process_message(self, event):
    """处理飞书消息的异步任务"""
    try:
        # 转换消息格式
        claude_input = convert_feishu_to_claude(event)

        # 调用 Claude
        claude = ClaudeClient(os.getenv("CLAUDE_API_KEY"))
        response = claude.complete(**claude_input)

        # 回复飞书
        reply_feishu_message(event["event"]["message"]["chat_id"],
            event["event"]["message"]["message_id"],
            response
        )
    except Exception as e:
        logger.error(f"Process message failed: {e}")
        self.retry(exc=e, countdown=60)