前言

在搭建自动化测试 Agent、业务 AI 智能体时，我们经常需要让自建服务通过飞书接收用户输入：

  • 文本指令（如：淘宝搜索商品、执行自动化用例）
  • 图片 / 文件上传（如：Excel 测试用例、TXT 配置文件）

本文从零到一完整实现：本地 Agent 服务 → 接收飞书消息回调 → 解析文本 / 图片 / 文件 → 过滤重复 / 历史消息所有代码可直接运行，附带全套排坑方案，适合搭建个人 / 企业智能体。

一、整体流程梳理

飞书消息 → 飞书开放平台服务器 → 内网穿透（ngrok） → 本地 FastAPI 服务 → Agent 获取结构化输入

核心链路：

  1. 用户在飞书发送消息 / 文件
  2. 飞书通过事件订阅推送 POST 回调
  3. ngrok 将公网流量转发到本地 8001 端口
  4. FastAPI 接收、验签、解析消息
  5. 去重 + 过滤历史消息，输出标准 JSON 给 Agent 使用

二、前期准备

2.1 飞书开放平台配置

进入飞书开放平台https://open.feishu.cn/，点击开发者后台，创建企业自建应用

也可以快速开发机器人，这边选择企业自建应用

  1. 「应用能力」→ 开启机器人
  2. 「权限管理」→ 直接导入即可，批量导入/导出权限,直接复制，黏贴导入 
    {
  "scopes": {
    "tenant": [
      "aily:file:read",
      "aily:file:write",
      "application:application.app_message_stats.overview:readonly",
      "application:application:self_manage",
      "application:bot.menu:write",
      "cardkit:card:write",
      "contact:contact.base:readonly",
      "contact:user.employee_id:readonly",
      "corehr:file:download",
      "docs:document.content:read",
      "event:ip_list",
      "im:chat",
      "im:chat.access_event.bot_p2p_chat:read",
      "im:chat.members:bot_access",
      "im:message",
      "im:message.group_at_msg:readonly",
      "im:message.group_msg",
      "im:message.p2p_msg:readonly",
      "im:message:readonly",
      "im:message:send_as_bot",
      "im:resource",
      "sheets:spreadsheet",
      "wiki:wiki:readonly"
    ],
    "user": [
      "aily:file:read",
      "aily:file:write",
      "im:chat.access_event.bot_p2p_chat:read"
    ]
  }
}
  3. 「事件与回调」→ 事件配置，添加事件：im.message.receive_v1（接收用户发给机器人的消息）
  4. 记录关键信息：在加密策略：记录你的Verification Token
  5. 权限管理开启：
    • im:message
    • im:message:read
    • im:message.file:read（文件必开）

2.2 内网穿透（ngrok）

飞书只能访问公网地址，本地服务必须做穿透：

  1. 注册https://ngrok.com/，下载 Windows 客户端
  2. 配置 authtoken： 
    ngrok config add-authtoken 你的token
  3. 映射本地端口： 
    ngrok http 8001
  4. 复制生成的公网 HTTPS 地址，填入飞书「事件订阅请求地址」： 
    https://xxx.ngrok-free.dev/feishu/receive

2.4 版本管理与发布

创建版本，填入版本号和更新说明，保存即可发布，这样你的飞书客户端就会收到发布成功的通知

2.4 Python 依赖安装

pip install fastapi uvicorn requests pandas openpyxl

三、核心代码实现

3.1 基础接收服务（URL 校验 + 消息接收 + 去重 + 过滤）

解决三大问题：

  • 飞书 URL 校验（challenge 返回）
  • 消息重复推送（飞书重试机制）
  • 历史 / 删除消息干扰（时间戳过滤）
from fastapi import FastAPI, Request
import json
import time

app = FastAPI()

# 飞书配置（从开放平台复制）
FEISHU_VERIFY_TOKEN = "你的Verification Token"

# 消息去重集合 + 过期时间（10分钟内消息才处理）
processed_event_ids = set()
EXPIRE_SECONDS = 600

@app.post("/feishu/receive")
async def receive_feishu_msg(request: Request):
    # 1. 获取原始消息体
    body = await request.body()
    raw_data = json.loads(body.decode("utf-8"))

    # 2. 飞书URL校验（配置回调时必过）
    if raw_data.get("type") == "url_verification":
        return {"challenge": raw_data.get("challenge")}

    try:
        # 基础字段抽取
        header = raw_data["header"]
        event = raw_data["event"]
        event_id = header["event_id"]       # 消息唯一ID
        msg_time = int(event["message"]["create_time"]) // 1000
        msg_type = event["message"]["message_type"]
        content = json.loads(event["message"]["content"])

        # 3. 消息去重：同一event_id只处理一次
        if event_id in processed_event_ids:
            return {"code": 200, "msg": "重复消息，已忽略"}

        # 4. 过滤历史消息（只处理10分钟内新消息）
        if time.time() - msg_time > EXPIRE_SECONDS:
            processed_event_ids.add(event_id)
            return {"code": 200, "msg": "历史消息，已过滤"}

        # 标记为已处理
        processed_event_ids.add(event_id)

        # 5. 多类型消息解析
        result = {}
        if msg_type == "text":
            result = {
                "type": "文本",
                "text": content.get("text", "")
            }
        elif msg_type == "image":
            result = {
                "type": "图片",
                "image_key": content.get("image_key", "")
            }
        elif msg_type == "file":
            result = {
                "type": "文件",
                "file_name": content.get("name", ""),
                "file_key": content.get("file_key", "")
            }
        else:
            result = {"type": "未知类型"}

        # 标准结构化输出（Agent直接使用）
        return {
            "code": 200,
            "msg": "success",
            "data": {
                "event_id": event_id,
                "message": result
            }
        }

    except Exception as e:
        return {"code": 500, "msg": f"解析失败：{str(e)}"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

四、进阶：接收文件并读取内容（TXT/Excel）

仅拿到file_key无法获取文件内容，需调用飞书接口下载并读取

from fastapi import FastAPI, Request
import json
import time
import requests
import pandas as pd

app = FastAPI()

# 飞书配置
FEISHU_VERIFY_TOKEN = "你的Verification Token"
FEISHU_APP_ID = "你的App ID"
FEISHU_APP_SECRET = "你的App Secret"

processed_event_ids = set()
EXPIRE_SECONDS = 600

# ========== 获取飞书接口凭证 ==========
def get_tenant_token():
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    resp = requests.post(url, json={
        "app_id": FEISHU_APP_ID,
        "app_secret": FEISHU_APP_SECRET
    })
    return resp.json()["tenant_access_token"]

# ========== 下载并读取飞书文件 ==========
def parse_file_content(file_key, file_name):
    token = get_tenant_token()
    headers = {"Authorization": f"Bearer {token}"}

    # 1. 获取文件下载链接
    info_resp = requests.get(
        f"https://open.feishu.cn/open-apis/im/v1/files/{file_key}",
        headers=headers
    )
    download_url = info_resp.json()["data"]["download_url"]

    # 2. 下载文件
    file_bytes = requests.get(download_url, headers=headers).content

    # 3. 根据类型解析
    try:
        if file_name.endswith(".txt"):
            return {"type": "txt", "content": file_bytes.decode("utf-8")}
        elif file_name.endswith(".xlsx"):
            df = pd.read_excel(pd.compat.BytesIO(file_bytes))
            return {"type": "excel", "content": df.to_dict(orient="records")}
        else:
            return {"type": "unknown", "content": "不支持该格式"}
    except:
        return {"type": "error", "content": "文件读取失败"}

# ========== 主接收接口 ==========
@app.post("/feishu/receive")
async def receive_feishu_msg(request: Request):
    body = await request.body()
    raw_data = json.loads(body.decode("utf-8"))

    if raw_data.get("type") == "url_verification":
        return {"challenge": raw_data.get("challenge")}

    try:
        event_id = raw_data["header"]["event_id"]
        msg_time = int(raw_data["event"]["message"]["create_time"])//1000
        msg_type = raw_data["event"]["message"]["message_type"]
        content = json.loads(raw_data["event"]["message"]["content"])

        # 去重 + 时间过滤
        if event_id in processed_event_ids:
            return {"code":200,"msg":"重复消息"}
        if time.time() - msg_time > EXPIRE_SECONDS:
            processed_event_ids.add(event_id)
            return {"code":200,"msg":"历史消息"}
        processed_event_ids.add(event_id)

        # 消息解析
        if msg_type == "text":
            data = {"type":"文本","text":content["text"]}
        elif msg_type == "file":
            file_content = parse_file_content(content["file_key"], content["name"])
            data = {
                "type":"文件",
                "name":content["name"],
                "file_content": file_content
            }
        else:
            data = {"type":msg_type}

        return {"code":200,"msg":"success","data":data}

    except Exception as e:
        return {"code":500,"msg":str(e)}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8001)

五、运行效果

代码运行之后：可以在飞书输入简单文字，后端会接收到你输入的文字

5.1 文本消息输入

飞书发送：你好：

{
  "code": 200,
  "msg": "success",
  "data": {
    "type": "文本",
    "text": "你好"
  }
}

5.2 文件输入（TXT/Excel）

上传test.txt，内容：Agent：

{
  "code": 200,
  "data": {
    "type": "文件",
    "name": "test.txt",
    "file_content": {
      "type": "txt",
      "content": "Agent"
    }
  }
}

六、高频问题排坑（实战必看）

6.1 验签失败（403 Forbidden）

  • 飞书 v2.0 事件使用SHA256+timestamp+nonce+body验签
  • 测试阶段可临时跳过验签，优先保证消息接收

6.2 一条消息重复打印多次

  • 原因：飞书未及时收到 200 响应，自动重试
  • 解决：使用event_id做内存去重（本文已实现）

6.3 收到很久以前的聊天记录

  • 原因：飞书会推送订阅前的历史消息
  • 解决：按时间戳过滤，只处理 10 分钟内消息

6.4 仅支持文本，图片 / 文件报错

  • 原因：未兼容image/file类型结构
  • 解决：按message_type分支解析

6.5 飞书回调配置失败

  • 必须使用HTTPS公网地址
  • 路由必须完全一致：/feishu/receive
  • 必须正确返回challenge
# python # AI
Logo
腾讯云开发者社区

腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。

更多推荐

Elasticsearch复杂数据类型终极指南：从入门到精通

Elasticsearch作为功能强大的搜索引擎，支持多种复杂数据类型，让开发者能够灵活处理各种结构化和非结构化数据。本文将带你全面了解Elasticsearch中的复杂数据类型，从基础概念到实际应用，助你轻松掌握数据建模的核心技巧。## 内部对象：构建层级化数据结构在Elasticsearch中，对象类型（Object）是最基础的复杂数据类型之一，用于表示具有嵌套关系的数据。例如，我们可

avatar 腾讯云开发者社区

终极指南：Flink SQL连接器版本管理从混乱到有序的升级之路

Apache Flink作为流处理领域的佼佼者，其SQL连接器的版本管理一直是开发者面临的核心挑战。本文将系统讲解Flink SQL连接器版本管理的最佳实践，帮助你轻松应对版本兼容性问题，实现从混乱到有序的升级之旅。## 连接器版本管理的常见痛点 😫在Flink应用开发中，连接器版本管理常常让开发者头疼不已。不同版本的连接器可能导致各种兼容性问题，例如API变更、功能差异甚至运行时错误。

avatar 腾讯云开发者社区

如何快速搭建Neon无服务器PostgreSQL：面向初学者的完整指南

Neon是一款革命性的无服务器PostgreSQL解决方案，它通过分离存储和计算层，实现了自动扩缩容、类代码式数据库分支以及零级扩展能力。本指南将帮助你从零开始搭建Neon开发环境，体验这款创新数据库的强大功能。## 准备工作：环境要求与依赖项在开始搭建Neon环境前，请确保你的系统满足以下要求：- Linux操作系统（推荐Ubuntu 20.04+或Debian 11+）- Git

avatar 腾讯云开发者社区