1. 项目概述为什么用Python操作飞书API是刚需最近在帮团队做自动化流程改造发现很多重复性的通知、数据同步和审批流转还在靠人工手动操作效率低不说还容易出错。比如每天要把销售数据从数据库导出来再手动粘贴到飞书多维表格里或者新用户注册后需要手动拉群、发欢迎消息。这些工作琐碎又耗时完全可以用代码自动化。飞书作为一款集成了IM、日历、文档、表格的协同办公平台其开放的API接口就是我们实现自动化的“金钥匙”。而Python凭借其简洁的语法和丰富的第三方库无疑是操作这些API最顺手、最高效的工具之一。这个项目就是围绕如何用Python这把“瑞士军刀”去灵活、稳定地操作飞书API解决实际办公场景中的痛点。无论是想自动发送日报、同步数据到多维表格、创建审批流程还是搭建一个能自动回复消息的机器人核心都离不开对飞书API的调用。整个过程听起来有点技术门槛但只要你跟着步骤走会发现它比想象中简单。接下来我会从零开始拆解整个流程包括环境准备、认证鉴权、核心接口调用以及那些官方文档里不会写的“坑”和技巧。2. 核心思路与工具选型构建稳健的请求链路操作任何API本质上就是按照其规定的“语言”协议和“格式”数据向指定的“地址”端点发送请求并处理返回的“答复”响应。对于飞书API我们需要重点关注几个核心环节身份认证、请求构造、错误处理和速率控制。2.1 身份认证获取访问凭证飞书API主要使用两种认证方式自建应用和企业自建应用。对于个人或小团队内部自动化我强烈推荐使用“企业自建应用”它权限可控无需应用上架审核流程更简单。创建应用登录 飞书开放平台 进入“开发者后台”创建一个新的“企业自建应用”。给你的应用起个名字比如“数据同步机器人”。获取凭证应用创建后在“凭证与基础信息”页面你会找到两个关键信息app_id: 应用的唯一标识。app_secret: 应用的密钥务必保密相当于密码。申请权限在“权限管理”页面根据你的需求为应用添加对应的权限。例如要发送消息就需要添加“以应用身份发送消息”、“获取用户发给机器人的单聊消息”等权限。添加后记得点击“申请线上发布”通常管理员秒批。获取tenant_access_token这是调用大多数API所需的令牌。你需要用app_id和app_secret去换取。这个令牌有效期通常为2小时需要定时刷新。这是整个流程的第一个关键点。注意app_secret一旦泄露他人可以冒充你的应用进行操作。千万不要把它硬编码在代码里提交到Git等公开仓库。一定要使用环境变量或配置文件并确保.gitignore排除了这些敏感文件。2.2 HTTP客户端选型requests库是不二之选Python中有urllib、httpx等HTTP库但对于飞书API这种标准的RESTful接口requests库以其极简的API和广泛的社区支持是绝大多数场景下的最佳选择。它的安装和使用都异常简单。pip install requests2.3 辅助工具让开发更高效JSON处理飞书API的请求和响应基本都是JSON格式。Python内置的json模块完全够用用于序列化json.dumps()和反序列化json.loads()数据。环境变量管理使用python-dotenv库来管理app_id和app_secret等敏感信息实现配置与代码分离。pip install python-dotenv日志记录使用内置的logging模块记录API调用情况、错误信息便于后期调试和监控。3. 实战第一步环境准备与基础封装理论说再多不如动手写一行代码。我们先搭建一个最基础、可复用的飞书API客户端类。3.1 项目结构与敏感信息管理首先建立清晰的项目目录结构feishu_bot/ ├── .env # 存储敏感信息务必加入.gitignore ├── config.py # 配置文件读取环境变量 ├── feishu_client.py # 核心的飞书API客户端类 ├── main.py # 主程序写业务逻辑 └── requirements.txt # 项目依赖在.env文件中存放你的凭证FEISHU_APP_IDcli_xxxxxx FEISHU_APP_SECRETxxxxxx-xxxxxx在config.py中安全地读取它们import os from dotenv import load_dotenv load_dotenv() # 加载.env文件中的环境变量 class Config: APP_ID os.getenv(FEISHU_APP_ID) APP_SECRET os.getenv(FEISHU_APP_SECRET) if not APP_ID or not APP_SECRET: raise ValueError(请在 .env 文件中配置 FEISHU_APP_ID 和 FEISHU_APP_SECRET)3.2 封装基础客户端类在feishu_client.py中我们创建一个FeishuClient类它负责管理token和发送请求。import requests import json import time import logging from config import Config logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class FeishuClient: def __init__(self, app_idNone, app_secretNone): self.app_id app_id or Config.APP_ID self.app_secret app_secret or Config.APP_SECRET self._tenant_access_token None self._token_expire_time 0 self.base_url https://open.feishu.cn/open-apis def _get_tenant_access_token(self): 内部方法获取或刷新tenant_access_token # 如果token存在且未过期直接返回 if self._tenant_access_token and time.time() self._token_expire_time: return self._tenant_access_token # 否则请求新的token url f{self.base_url}/auth/v3/tenant_access_token/internal headers {Content-Type: application/json; charsetutf-8} payload { app_id: self.app_id, app_secret: self.app_secret } try: response requests.post(url, headersheaders, jsonpayload, timeout10) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 result response.json() if result.get(code) 0: self._tenant_access_token result[tenant_access_token] # 设置过期时间通常有效期7200秒这里预留60秒缓冲 self._token_expire_time time.time() result.get(expire, 7200) - 60 logger.info(Tenant access token 获取成功) return self._tenant_access_token else: logger.error(f获取token失败: {result}) raise Exception(fAuth Error: {result.get(msg)}) except requests.exceptions.RequestException as e: logger.error(f请求token时网络错误: {e}) raise except json.JSONDecodeError as e: logger.error(f解析token响应JSON失败: {e}) raise def request(self, method, endpoint, **kwargs): 统一的请求方法自动处理token和基础错误 token self._get_tenant_access_token() url f{self.base_url}{endpoint} headers { Authorization: fBearer {token}, Content-Type: application/json; charsetutf-8, } # 如果调用者传入了headers则合并但Authorization优先级最高 if headers in kwargs: headers.update(kwargs.pop(headers)) logger.debug(f请求飞书API: {method} {url}) try: response requests.request(method, url, headersheaders, **kwargs) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: # 这里可以细化处理不同的HTTP状态码比如429限流、401token失效等 error_msg fHTTP错误: {e.response.status_code} try: error_detail e.response.json() error_msg f, 响应: {error_detail} # 特别处理token失效的情况清空token以便下次重试 if e.response.status_code 401: self._tenant_access_token None logger.warning(Token可能已失效已清空缓存) except: error_msg f, 响应文本: {e.response.text} logger.error(error_msg) raise except requests.exceptions.RequestException as e: logger.error(f网络请求异常: {e}) raise这个FeishuClient类已经具备了最核心的能力自动管理token生命周期提供统一的、带错误处理的请求方法。后续所有具体的API操作都将基于这个request方法展开。4. 核心接口调用实战从发送消息到操作表格有了稳固的基础设施我们就可以开始实现具体的业务功能了。飞书API功能模块非常多这里挑几个最常用、最典型的场景来详细拆解。4.1 发送消息到个人或群聊发送消息是机器人最基本的功能。飞书支持文本、富文本post、图片、文件、群卡片等多种消息类型。1. 获取会话IDchat_id或open_id要发消息首先要知道发给谁。对于群聊你需要群的chat_id对于个人你需要用户的open_id或user_id。获取群chat_id最方便的方法是将机器人拉入目标群机器人在群里发言后可以通过“获取机器人所在的群列表”API来找到这个群的chat_id。获取用户open_id在开放平台的应用后台“权限管理” - “开通权限”中申请“获取用户 user ID”等权限。然后可以通过“获取用户列表”或“通过手机号/邮箱获取用户ID”等API来查询。2. 发送文本消息示例假设我们已经拿到了一个群的chat_id发送一条简单的文本消息class FeishuMessage(FeishuClient): def send_text(self, receive_id_typechat_id, receive_idNone, contentNone): 发送文本消息 :param receive_id_type: 接收者ID类型open_id, user_id, email, chat_id :param receive_id: 接收者的ID :param content: 文本内容 if not content or not receive_id: raise ValueError(receive_id 和 content 不能为空) endpoint /im/v1/messages params {receive_id_type: receive_id_type} payload { receive_id: receive_id, msg_type: text, content: json.dumps({text: content}) # 注意content需要是JSON字符串 } result self.request(POST, endpoint, paramsparams, jsonpayload) if result.get(code) 0: message_id result.get(data, {}).get(message_id) logger.info(f消息发送成功message_id: {message_id}) return message_id else: logger.error(f消息发送失败: {result}) return None # 使用示例 if __name__ __main__: client FeishuMessage() # 替换为你的群chat_id chat_id oc_xxxxxx client.send_text(receive_id_typechat_id, receive_idchat_id, content大家好这是来自Python机器人的测试消息)3. 发送富文本Post消息文本消息太单调Post消息可以支持标题、加粗、链接、图片人等复杂排版信息呈现更清晰。def send_post(self, receive_id_type, receive_id, title, content): 发送富文本Post消息 :param title: 帖子标题 :param content: 帖子内容是一个列表每个元素代表一行或一个段落。 格式参考飞书文档例如[{tag: text, text: Hello}] endpoint /im/v1/messages params {receive_id_type: receive_id_type} # 构建Post消息结构 post_content { zh_cn: { # 语言zh_cn表示简体中文 title: title, content: content } } payload { receive_id: receive_id, msg_type: post, content: json.dumps(post_content) } return self.request(POST, endpoint, paramsparams, jsonpayload) # 使用示例发送一个带标题和分段的内容 post_content [ [ # 第一行 {tag: text, text: 今日项目日报, style: [{bold: True}]}, {tag: a, text: 查看详情, href: https://your-domain.com/report} ], [ # 第二行 {tag: text, text: 进度: }, {tag: text, text: 90%, style: [{color: green}]} ], [ # 第三行 {tag: at, user_id: user_id_xxx} # 某人 ] ] client.send_post(chat_id, chat_id, 项目更新通知, post_content)实操心得消息内容中的content字段必须是一个JSON字符串这是新手最容易踩的坑。直接传Python字典会报错。务必使用json.dumps()进行转换。另外Post消息的结构比较复杂建议先在飞书开放平台的“消息格式”文档里找到对应模板再修改成自己的内容。4.2 操作飞书多维表格多维表格是飞书里非常强大的数据管理工具通过API可以自动化地进行增删改查实现外部系统与表格的数据同步。1. 准备工作获取app_token和table_id操作某个具体的多维表格前你需要两个IDapp_token: 多维表格本身的标识。在浏览器中打开你的多维表格地址栏URL中base参数后面的值就是app_token。例如...basexxxxxxxxxx。table_id: 表格内某个具体子表的标识。在表格页面点击右上角“...” - “复制链接”链接中table参数后面的值就是table_id。2. 新增记录示例假设我们有一个“任务清单”表格有“任务名”文本和“截止日期”日期两个字段。class FeishuBitable(FeishuClient): def add_record(self, app_token, table_id, fields): 向多维表格添加一条记录 :param app_token: 多维表格的token :param table_id: 表格ID :param fields: 记录字段字典键为字段名值为字段值 :return: 新增记录的ID endpoint f/bitable/v1/apps/{app_token}/tables/{table_id}/records payload { fields: fields } result self.request(POST, endpoint, jsonpayload) if result.get(code) 0: record_id result.get(data, {}).get(record, {}).get(record_id) logger.info(f记录添加成功record_id: {record_id}) return record_id else: logger.error(f添加记录失败: {result}) return None # 使用示例 if __name__ __main__: client FeishuBitable() app_token xxxxxxxxxx table_id tblxxxxxxxxxx # 注意字段值的格式必须符合字段类型 new_task { 任务名: 编写API文档, 截止日期: 1735660800000 # 日期字段需要传时间戳毫秒 } record_id client.add_record(app_token, table_id, new_task)3. 查询记录与分页处理表格数据多了查询时一定要处理分页。飞书API的列表接口通常支持page_size和page_token参数。def list_records(self, app_token, table_id, page_size100, filter_formulaNone): 列出表格中的所有记录自动处理分页 :param filter_formula: 过滤公式例如 CurrentValue.[状态] 进行中 :return: 所有记录的列表 all_records [] page_token None while True: endpoint f/bitable/v1/apps/{app_token}/tables/{table_id}/records params { page_size: page_size, } if page_token: params[page_token] page_token if filter_formula: params[filter] filter_formula result self.request(GET, endpoint, paramsparams) if result.get(code) ! 0: logger.error(f查询记录失败: {result}) break data result.get(data, {}) items data.get(items, []) all_records.extend(items) # 检查是否有下一页 has_more data.get(has_more, False) page_token data.get(page_token) if not has_more or not page_token: break # 建议在循环中加一个小延迟避免请求过快 time.sleep(0.1) logger.info(f共查询到 {len(all_records)} 条记录) return all_records注意事项多维表格API的字段值格式非常严格。文本、数字、单选等类型直接传值即可但日期字段需要传毫秒级时间戳人员字段需要传包含id和type的对象如{id: ou_xxx, type: User}附件字段需要先上传文件获取file_token。务必在飞书开放平台查阅对应字段类型的API文档说明。一个常见的错误400 type must be in [enabled, disabled, auto]往往就是因为传入的字段值格式不符合预期系统无法解析。4.3 处理机器人事件与回调如果你希望机器人能响应用户的消息或点击卡片按钮就需要配置事件订阅。这比主动发送消息复杂一些涉及搭建一个能接收HTTP POST请求的Webhook服务器。核心流程配置事件订阅在开放平台应用后台的“事件订阅”中设置请求网址URL你的服务器公网地址并订阅所需事件如“接收消息”、“消息已读”等。验证URL飞书服务器会向你的URL发送一个带challenge参数的GET请求你需要原样返回这个challenge值以验证URL归属。处理事件验证通过后飞书会将用户事件如消息以POST请求的形式推送到你的URL你需要解析加密数据并做出响应。简化示例使用Flask框架# event_handler.py from flask import Flask, request, jsonify import json from feishu_client import FeishuClient # 导入之前封装的客户端 app Flask(__name__) client FeishuClient() # 1. 验证URL有效性 app.route(/webhook/feishu, methods[GET]) def verify(): challenge request.args.get(challenge) if challenge: return jsonify({challenge: challenge}) return Error, 400 # 2. 处理事件推送 app.route(/webhook/feishu, methods[POST]) def handle_event(): data request.json # 这里应该包含解密逻辑如果启用了加密此处简化 event_type data.get(header, {}).get(event_type) event data.get(event, {}) if event_type im.message.receive_v1: # 收到消息事件 message event.get(message, {}) msg_type message.get(message_type) content json.loads(message.get(content, {})) sender event.get(sender, {}) if msg_type text: text_content content.get(text, ) # 判断是否是机器人的消息 if fat_{Config.APP_ID} in text_content: # 提取纯文本内容去除信息 pure_text text_content.replace(fat_{Config.APP_ID}, ).strip() # 调用之前封装的方法回复 chat_id message.get(chat_id) reply_content f你好我收到了你的消息{pure_text} client.send_text(chat_id, chat_id, reply_content) return jsonify({code: 0}) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)这个示例省略了数据加密解密和签名验证的关键步骤。在生产环境中你必须在开放平台配置“加密密钥”并在代码中实现对应的解密算法否则无法接收到真实消息内容。这是事件订阅模式最大的挑战也是安全性的保障。5. 高级技巧与避坑指南在实际开发和运维中会遇到很多官方文档没有明确说明的问题。这里分享几个我踩过坑后总结的经验。5.1 错误处理与重试机制网络请求不可能100%成功。我们必须为各种异常设计健壮的处理逻辑。import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class RobustFeishuClient(FeishuClient): def __init__(self, app_idNone, app_secretNone): super().__init__(app_id, app_secret) # 为requests session配置重试策略 self.session requests.Session() retry_strategy Retry( total3, # 最大重试次数 backoff_factor1, # 重试等待时间因子 status_forcelist[429, 500, 502, 503, 504], # 遇到这些状态码重试 allowed_methods[GET, POST, PUT, DELETE] # 只对这些方法重试 ) adapter HTTPAdapter(max_retriesretry_strategy) self.session.mount(https://, adapter) self.session.mount(http://, adapter) def request(self, method, endpoint, **kwargs): # 使用配置了重试的session token self._get_tenant_access_token() url f{self.base_url}{endpoint} headers { Authorization: fBearer {token}, Content-Type: application/json; charsetutf-8, } if headers in kwargs: headers.update(kwargs.pop(headers)) logger.debug(f请求飞书API (带重试): {method} {url}) try: # 使用self.session.request response self.session.request(method, url, headersheaders, **kwargs) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: # 细化错误处理 error_code e.response.status_code if error_code 429: logger.warning(触发API速率限制建议稍后重试或检查调用频率) # 可以在这里读取响应头的 Retry-After 字段进行精确等待 elif error_code 400: # 仔细分析400错误的返回体常见于参数错误 error_body e.response.json() logger.error(f请求参数错误 (400): {error_body}) if maximum context length in str(error_body): logger.error(错误提示请求内容过长请检查输入数据大小。) raise5.2 应对API速率限制飞书API有严格的调用频率限制QPS。在_get_tenant_access_token等高频调用的接口上一定要做好本地缓存避免重复请求。对于业务接口如果批量操作大量数据如导入成千上万条表格记录必须在代码中主动加入延迟。import time def batch_add_records(client, app_token, table_id, records_list): 批量添加记录并主动控制速率 success_count 0 fail_count 0 for i, record_fields in enumerate(records_list): try: client.add_record(app_token, table_id, record_fields) success_count 1 except Exception as e: logger.error(f添加第{i1}条记录失败: {e}) fail_count 1 # 每处理10条记录暂停1秒避免触发限流 if (i 1) % 10 0: time.sleep(1) # 或者更精细地根据API的剩余配额如果响应头有提供来动态调整 logger.info(f批量导入完成成功{success_count}条失败{fail_count}条)5.3 文件上传与下载飞书消息支持发送图片、文件多维表格也支持附件字段。文件操作通常分两步调用/drive/v1/files/upload_all接口上传文件获取file_token。发送消息时在content中引用这个file_token或在多维表格字段值中填入[{file_token: xxxxx}]。上传文件时需要注意接口对文件大小、格式的限制并且要正确设置Content-Type为multipart/form-data。5.4 调试与日志完善的日志是快速定位问题的关键。除了记录INFO级别的操作更要将请求的URL、关键参数、响应状态码和错误信息记录在DEBUG或ERROR级别。建议使用像loguru这样更友好的日志库并配置日志滚动归档方便后期排查历史问题。6. 常见问题排查与解决方案实录在实际对接中你几乎一定会遇到下面这些问题。我把它们和解决方案整理成了表格方便快速查阅。问题现象可能原因排查步骤与解决方案400错误提示type must be in [enabled, disabled, auto]请求体或查询参数中某个字段的值不符合API枚举要求。1. 仔细检查报错接口的官方文档确认type字段允许的值。2. 核对代码中传入的值是否完全一致大小写敏感。3. 使用print或日志输出完整的请求体与文档示例对比。400错误提示this models maximum context length is ... tokens请求内容如消息文本、表格单元格内容过长超过了接口限制。1. 此错误常见于AI相关接口检查发送的文本是否超长。2. 对于普通消息飞书单条文本消息有限制需分段发送。3. 对于多维表格检查单个单元格内容是否过多。401认证失败tenant_access_token无效或已过期。1. 检查app_id和app_secret是否正确。2. 检查客户端代码的token缓存和刷新逻辑是否正常。3. 在开放平台后台检查应用是否已被停用。403无权限应用没有调用该接口的权限。1. 登录开放平台在“权限管理”中查看是否已添加对应权限。2. 添加权限后是否已点击“申请发布”并获管理员批准。3. 部分权限如获取部门下所有用户需要申请更高级别的权限。429请求过于频繁触发了飞书API的速率限制。1. 立即停止当前批量请求。2. 在代码中实现指数退避算法的重试机制。3. 检查业务逻辑优化代码减少不必要的API调用如缓存用户信息。4. 查看响应头是否有Retry-After按其指示的时间等待。消息发送成功但收不到1. 机器人不在该群。2. 发送给了错误的chat_id或open_id。3. 群或用户已屏蔽机器人。1. 确认机器人已加入目标群。2. 使用“获取群列表”API确认使用的chat_id是否正确。3. 尝试向其他群或用户发送以排除屏蔽可能。多维表格操作失败字段值不生效字段值格式与字段类型不匹配。1.这是最高频的错误。在表格页面查看字段的确切类型。2. 对照官方文档检查传入的数据格式-日期毫秒时间戳整数-人员{id: “user_id”, “type”: “User”}对象-单选选项ID字符串-附件[{file_token: “xxx”}]列表事件订阅URL验证不通过1. 服务器未正确响应challenge。2. 网络问题导致飞书无法访问你的URL。3. URL使用了HTTPS但证书有问题。1. 确保你的服务器公网可访问且/webhook路径的GET请求能返回{“challenge”: “xxx”}。2. 使用curl或在线工具手动测试你的URL。3. 本地开发可使用内网穿透工具如ngrok提供临时公网地址。收到事件但解密失败未在代码中实现解密逻辑或加密密钥配置错误。1. 在开放平台“事件订阅”中确认已设置“加密密钥”。2. 在代码中严格按照飞书官方提供的加解密SDK或示例处理encrypt字段。最后再分享一个我自己的体会不要重复造轮子。飞书官方为Python提供了SDK (lark-oapi)它封装了认证、请求、加解密等几乎所有底层细节。对于生产环境尤其是需要处理事件订阅的复杂应用直接使用官方SDK是更稳定、更高效的选择。本文从零开始手写是为了让大家彻底理解其原理但在实际项目中评估后引入官方SDK往往能事半功倍。 Python自动化飞书API：从零构建机器人，实现消息与表格操作
1. 项目概述为什么用Python操作飞书API是刚需最近在帮团队做自动化流程改造发现很多重复性的通知、数据同步和审批流转还在靠人工手动操作效率低不说还容易出错。比如每天要把销售数据从数据库导出来再手动粘贴到飞书多维表格里或者新用户注册后需要手动拉群、发欢迎消息。这些工作琐碎又耗时完全可以用代码自动化。飞书作为一款集成了IM、日历、文档、表格的协同办公平台其开放的API接口就是我们实现自动化的“金钥匙”。而Python凭借其简洁的语法和丰富的第三方库无疑是操作这些API最顺手、最高效的工具之一。这个项目就是围绕如何用Python这把“瑞士军刀”去灵活、稳定地操作飞书API解决实际办公场景中的痛点。无论是想自动发送日报、同步数据到多维表格、创建审批流程还是搭建一个能自动回复消息的机器人核心都离不开对飞书API的调用。整个过程听起来有点技术门槛但只要你跟着步骤走会发现它比想象中简单。接下来我会从零开始拆解整个流程包括环境准备、认证鉴权、核心接口调用以及那些官方文档里不会写的“坑”和技巧。2. 核心思路与工具选型构建稳健的请求链路操作任何API本质上就是按照其规定的“语言”协议和“格式”数据向指定的“地址”端点发送请求并处理返回的“答复”响应。对于飞书API我们需要重点关注几个核心环节身份认证、请求构造、错误处理和速率控制。2.1 身份认证获取访问凭证飞书API主要使用两种认证方式自建应用和企业自建应用。对于个人或小团队内部自动化我强烈推荐使用“企业自建应用”它权限可控无需应用上架审核流程更简单。创建应用登录 飞书开放平台 进入“开发者后台”创建一个新的“企业自建应用”。给你的应用起个名字比如“数据同步机器人”。获取凭证应用创建后在“凭证与基础信息”页面你会找到两个关键信息app_id: 应用的唯一标识。app_secret: 应用的密钥务必保密相当于密码。申请权限在“权限管理”页面根据你的需求为应用添加对应的权限。例如要发送消息就需要添加“以应用身份发送消息”、“获取用户发给机器人的单聊消息”等权限。添加后记得点击“申请线上发布”通常管理员秒批。获取tenant_access_token这是调用大多数API所需的令牌。你需要用app_id和app_secret去换取。这个令牌有效期通常为2小时需要定时刷新。这是整个流程的第一个关键点。注意app_secret一旦泄露他人可以冒充你的应用进行操作。千万不要把它硬编码在代码里提交到Git等公开仓库。一定要使用环境变量或配置文件并确保.gitignore排除了这些敏感文件。2.2 HTTP客户端选型requests库是不二之选Python中有urllib、httpx等HTTP库但对于飞书API这种标准的RESTful接口requests库以其极简的API和广泛的社区支持是绝大多数场景下的最佳选择。它的安装和使用都异常简单。pip install requests2.3 辅助工具让开发更高效JSON处理飞书API的请求和响应基本都是JSON格式。Python内置的json模块完全够用用于序列化json.dumps()和反序列化json.loads()数据。环境变量管理使用python-dotenv库来管理app_id和app_secret等敏感信息实现配置与代码分离。pip install python-dotenv日志记录使用内置的logging模块记录API调用情况、错误信息便于后期调试和监控。3. 实战第一步环境准备与基础封装理论说再多不如动手写一行代码。我们先搭建一个最基础、可复用的飞书API客户端类。3.1 项目结构与敏感信息管理首先建立清晰的项目目录结构feishu_bot/ ├── .env # 存储敏感信息务必加入.gitignore ├── config.py # 配置文件读取环境变量 ├── feishu_client.py # 核心的飞书API客户端类 ├── main.py # 主程序写业务逻辑 └── requirements.txt # 项目依赖在.env文件中存放你的凭证FEISHU_APP_IDcli_xxxxxx FEISHU_APP_SECRETxxxxxx-xxxxxx在config.py中安全地读取它们import os from dotenv import load_dotenv load_dotenv() # 加载.env文件中的环境变量 class Config: APP_ID os.getenv(FEISHU_APP_ID) APP_SECRET os.getenv(FEISHU_APP_SECRET) if not APP_ID or not APP_SECRET: raise ValueError(请在 .env 文件中配置 FEISHU_APP_ID 和 FEISHU_APP_SECRET)3.2 封装基础客户端类在feishu_client.py中我们创建一个FeishuClient类它负责管理token和发送请求。import requests import json import time import logging from config import Config logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class FeishuClient: def __init__(self, app_idNone, app_secretNone): self.app_id app_id or Config.APP_ID self.app_secret app_secret or Config.APP_SECRET self._tenant_access_token None self._token_expire_time 0 self.base_url https://open.feishu.cn/open-apis def _get_tenant_access_token(self): 内部方法获取或刷新tenant_access_token # 如果token存在且未过期直接返回 if self._tenant_access_token and time.time() self._token_expire_time: return self._tenant_access_token # 否则请求新的token url f{self.base_url}/auth/v3/tenant_access_token/internal headers {Content-Type: application/json; charsetutf-8} payload { app_id: self.app_id, app_secret: self.app_secret } try: response requests.post(url, headersheaders, jsonpayload, timeout10) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 result response.json() if result.get(code) 0: self._tenant_access_token result[tenant_access_token] # 设置过期时间通常有效期7200秒这里预留60秒缓冲 self._token_expire_time time.time() result.get(expire, 7200) - 60 logger.info(Tenant access token 获取成功) return self._tenant_access_token else: logger.error(f获取token失败: {result}) raise Exception(fAuth Error: {result.get(msg)}) except requests.exceptions.RequestException as e: logger.error(f请求token时网络错误: {e}) raise except json.JSONDecodeError as e: logger.error(f解析token响应JSON失败: {e}) raise def request(self, method, endpoint, **kwargs): 统一的请求方法自动处理token和基础错误 token self._get_tenant_access_token() url f{self.base_url}{endpoint} headers { Authorization: fBearer {token}, Content-Type: application/json; charsetutf-8, } # 如果调用者传入了headers则合并但Authorization优先级最高 if headers in kwargs: headers.update(kwargs.pop(headers)) logger.debug(f请求飞书API: {method} {url}) try: response requests.request(method, url, headersheaders, **kwargs) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: # 这里可以细化处理不同的HTTP状态码比如429限流、401token失效等 error_msg fHTTP错误: {e.response.status_code} try: error_detail e.response.json() error_msg f, 响应: {error_detail} # 特别处理token失效的情况清空token以便下次重试 if e.response.status_code 401: self._tenant_access_token None logger.warning(Token可能已失效已清空缓存) except: error_msg f, 响应文本: {e.response.text} logger.error(error_msg) raise except requests.exceptions.RequestException as e: logger.error(f网络请求异常: {e}) raise这个FeishuClient类已经具备了最核心的能力自动管理token生命周期提供统一的、带错误处理的请求方法。后续所有具体的API操作都将基于这个request方法展开。4. 核心接口调用实战从发送消息到操作表格有了稳固的基础设施我们就可以开始实现具体的业务功能了。飞书API功能模块非常多这里挑几个最常用、最典型的场景来详细拆解。4.1 发送消息到个人或群聊发送消息是机器人最基本的功能。飞书支持文本、富文本post、图片、文件、群卡片等多种消息类型。1. 获取会话IDchat_id或open_id要发消息首先要知道发给谁。对于群聊你需要群的chat_id对于个人你需要用户的open_id或user_id。获取群chat_id最方便的方法是将机器人拉入目标群机器人在群里发言后可以通过“获取机器人所在的群列表”API来找到这个群的chat_id。获取用户open_id在开放平台的应用后台“权限管理” - “开通权限”中申请“获取用户 user ID”等权限。然后可以通过“获取用户列表”或“通过手机号/邮箱获取用户ID”等API来查询。2. 发送文本消息示例假设我们已经拿到了一个群的chat_id发送一条简单的文本消息class FeishuMessage(FeishuClient): def send_text(self, receive_id_typechat_id, receive_idNone, contentNone): 发送文本消息 :param receive_id_type: 接收者ID类型open_id, user_id, email, chat_id :param receive_id: 接收者的ID :param content: 文本内容 if not content or not receive_id: raise ValueError(receive_id 和 content 不能为空) endpoint /im/v1/messages params {receive_id_type: receive_id_type} payload { receive_id: receive_id, msg_type: text, content: json.dumps({text: content}) # 注意content需要是JSON字符串 } result self.request(POST, endpoint, paramsparams, jsonpayload) if result.get(code) 0: message_id result.get(data, {}).get(message_id) logger.info(f消息发送成功message_id: {message_id}) return message_id else: logger.error(f消息发送失败: {result}) return None # 使用示例 if __name__ __main__: client FeishuMessage() # 替换为你的群chat_id chat_id oc_xxxxxx client.send_text(receive_id_typechat_id, receive_idchat_id, content大家好这是来自Python机器人的测试消息)3. 发送富文本Post消息文本消息太单调Post消息可以支持标题、加粗、链接、图片人等复杂排版信息呈现更清晰。def send_post(self, receive_id_type, receive_id, title, content): 发送富文本Post消息 :param title: 帖子标题 :param content: 帖子内容是一个列表每个元素代表一行或一个段落。 格式参考飞书文档例如[{tag: text, text: Hello}] endpoint /im/v1/messages params {receive_id_type: receive_id_type} # 构建Post消息结构 post_content { zh_cn: { # 语言zh_cn表示简体中文 title: title, content: content } } payload { receive_id: receive_id, msg_type: post, content: json.dumps(post_content) } return self.request(POST, endpoint, paramsparams, jsonpayload) # 使用示例发送一个带标题和分段的内容 post_content [ [ # 第一行 {tag: text, text: 今日项目日报, style: [{bold: True}]}, {tag: a, text: 查看详情, href: https://your-domain.com/report} ], [ # 第二行 {tag: text, text: 进度: }, {tag: text, text: 90%, style: [{color: green}]} ], [ # 第三行 {tag: at, user_id: user_id_xxx} # 某人 ] ] client.send_post(chat_id, chat_id, 项目更新通知, post_content)实操心得消息内容中的content字段必须是一个JSON字符串这是新手最容易踩的坑。直接传Python字典会报错。务必使用json.dumps()进行转换。另外Post消息的结构比较复杂建议先在飞书开放平台的“消息格式”文档里找到对应模板再修改成自己的内容。4.2 操作飞书多维表格多维表格是飞书里非常强大的数据管理工具通过API可以自动化地进行增删改查实现外部系统与表格的数据同步。1. 准备工作获取app_token和table_id操作某个具体的多维表格前你需要两个IDapp_token: 多维表格本身的标识。在浏览器中打开你的多维表格地址栏URL中base参数后面的值就是app_token。例如...basexxxxxxxxxx。table_id: 表格内某个具体子表的标识。在表格页面点击右上角“...” - “复制链接”链接中table参数后面的值就是table_id。2. 新增记录示例假设我们有一个“任务清单”表格有“任务名”文本和“截止日期”日期两个字段。class FeishuBitable(FeishuClient): def add_record(self, app_token, table_id, fields): 向多维表格添加一条记录 :param app_token: 多维表格的token :param table_id: 表格ID :param fields: 记录字段字典键为字段名值为字段值 :return: 新增记录的ID endpoint f/bitable/v1/apps/{app_token}/tables/{table_id}/records payload { fields: fields } result self.request(POST, endpoint, jsonpayload) if result.get(code) 0: record_id result.get(data, {}).get(record, {}).get(record_id) logger.info(f记录添加成功record_id: {record_id}) return record_id else: logger.error(f添加记录失败: {result}) return None # 使用示例 if __name__ __main__: client FeishuBitable() app_token xxxxxxxxxx table_id tblxxxxxxxxxx # 注意字段值的格式必须符合字段类型 new_task { 任务名: 编写API文档, 截止日期: 1735660800000 # 日期字段需要传时间戳毫秒 } record_id client.add_record(app_token, table_id, new_task)3. 查询记录与分页处理表格数据多了查询时一定要处理分页。飞书API的列表接口通常支持page_size和page_token参数。def list_records(self, app_token, table_id, page_size100, filter_formulaNone): 列出表格中的所有记录自动处理分页 :param filter_formula: 过滤公式例如 CurrentValue.[状态] 进行中 :return: 所有记录的列表 all_records [] page_token None while True: endpoint f/bitable/v1/apps/{app_token}/tables/{table_id}/records params { page_size: page_size, } if page_token: params[page_token] page_token if filter_formula: params[filter] filter_formula result self.request(GET, endpoint, paramsparams) if result.get(code) ! 0: logger.error(f查询记录失败: {result}) break data result.get(data, {}) items data.get(items, []) all_records.extend(items) # 检查是否有下一页 has_more data.get(has_more, False) page_token data.get(page_token) if not has_more or not page_token: break # 建议在循环中加一个小延迟避免请求过快 time.sleep(0.1) logger.info(f共查询到 {len(all_records)} 条记录) return all_records注意事项多维表格API的字段值格式非常严格。文本、数字、单选等类型直接传值即可但日期字段需要传毫秒级时间戳人员字段需要传包含id和type的对象如{id: ou_xxx, type: User}附件字段需要先上传文件获取file_token。务必在飞书开放平台查阅对应字段类型的API文档说明。一个常见的错误400 type must be in [enabled, disabled, auto]往往就是因为传入的字段值格式不符合预期系统无法解析。4.3 处理机器人事件与回调如果你希望机器人能响应用户的消息或点击卡片按钮就需要配置事件订阅。这比主动发送消息复杂一些涉及搭建一个能接收HTTP POST请求的Webhook服务器。核心流程配置事件订阅在开放平台应用后台的“事件订阅”中设置请求网址URL你的服务器公网地址并订阅所需事件如“接收消息”、“消息已读”等。验证URL飞书服务器会向你的URL发送一个带challenge参数的GET请求你需要原样返回这个challenge值以验证URL归属。处理事件验证通过后飞书会将用户事件如消息以POST请求的形式推送到你的URL你需要解析加密数据并做出响应。简化示例使用Flask框架# event_handler.py from flask import Flask, request, jsonify import json from feishu_client import FeishuClient # 导入之前封装的客户端 app Flask(__name__) client FeishuClient() # 1. 验证URL有效性 app.route(/webhook/feishu, methods[GET]) def verify(): challenge request.args.get(challenge) if challenge: return jsonify({challenge: challenge}) return Error, 400 # 2. 处理事件推送 app.route(/webhook/feishu, methods[POST]) def handle_event(): data request.json # 这里应该包含解密逻辑如果启用了加密此处简化 event_type data.get(header, {}).get(event_type) event data.get(event, {}) if event_type im.message.receive_v1: # 收到消息事件 message event.get(message, {}) msg_type message.get(message_type) content json.loads(message.get(content, {})) sender event.get(sender, {}) if msg_type text: text_content content.get(text, ) # 判断是否是机器人的消息 if fat_{Config.APP_ID} in text_content: # 提取纯文本内容去除信息 pure_text text_content.replace(fat_{Config.APP_ID}, ).strip() # 调用之前封装的方法回复 chat_id message.get(chat_id) reply_content f你好我收到了你的消息{pure_text} client.send_text(chat_id, chat_id, reply_content) return jsonify({code: 0}) if __name__ __main__: app.run(host0.0.0.0, port5000, debugTrue)这个示例省略了数据加密解密和签名验证的关键步骤。在生产环境中你必须在开放平台配置“加密密钥”并在代码中实现对应的解密算法否则无法接收到真实消息内容。这是事件订阅模式最大的挑战也是安全性的保障。5. 高级技巧与避坑指南在实际开发和运维中会遇到很多官方文档没有明确说明的问题。这里分享几个我踩过坑后总结的经验。5.1 错误处理与重试机制网络请求不可能100%成功。我们必须为各种异常设计健壮的处理逻辑。import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry class RobustFeishuClient(FeishuClient): def __init__(self, app_idNone, app_secretNone): super().__init__(app_id, app_secret) # 为requests session配置重试策略 self.session requests.Session() retry_strategy Retry( total3, # 最大重试次数 backoff_factor1, # 重试等待时间因子 status_forcelist[429, 500, 502, 503, 504], # 遇到这些状态码重试 allowed_methods[GET, POST, PUT, DELETE] # 只对这些方法重试 ) adapter HTTPAdapter(max_retriesretry_strategy) self.session.mount(https://, adapter) self.session.mount(http://, adapter) def request(self, method, endpoint, **kwargs): # 使用配置了重试的session token self._get_tenant_access_token() url f{self.base_url}{endpoint} headers { Authorization: fBearer {token}, Content-Type: application/json; charsetutf-8, } if headers in kwargs: headers.update(kwargs.pop(headers)) logger.debug(f请求飞书API (带重试): {method} {url}) try: # 使用self.session.request response self.session.request(method, url, headersheaders, **kwargs) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: # 细化错误处理 error_code e.response.status_code if error_code 429: logger.warning(触发API速率限制建议稍后重试或检查调用频率) # 可以在这里读取响应头的 Retry-After 字段进行精确等待 elif error_code 400: # 仔细分析400错误的返回体常见于参数错误 error_body e.response.json() logger.error(f请求参数错误 (400): {error_body}) if maximum context length in str(error_body): logger.error(错误提示请求内容过长请检查输入数据大小。) raise5.2 应对API速率限制飞书API有严格的调用频率限制QPS。在_get_tenant_access_token等高频调用的接口上一定要做好本地缓存避免重复请求。对于业务接口如果批量操作大量数据如导入成千上万条表格记录必须在代码中主动加入延迟。import time def batch_add_records(client, app_token, table_id, records_list): 批量添加记录并主动控制速率 success_count 0 fail_count 0 for i, record_fields in enumerate(records_list): try: client.add_record(app_token, table_id, record_fields) success_count 1 except Exception as e: logger.error(f添加第{i1}条记录失败: {e}) fail_count 1 # 每处理10条记录暂停1秒避免触发限流 if (i 1) % 10 0: time.sleep(1) # 或者更精细地根据API的剩余配额如果响应头有提供来动态调整 logger.info(f批量导入完成成功{success_count}条失败{fail_count}条)5.3 文件上传与下载飞书消息支持发送图片、文件多维表格也支持附件字段。文件操作通常分两步调用/drive/v1/files/upload_all接口上传文件获取file_token。发送消息时在content中引用这个file_token或在多维表格字段值中填入[{file_token: xxxxx}]。上传文件时需要注意接口对文件大小、格式的限制并且要正确设置Content-Type为multipart/form-data。5.4 调试与日志完善的日志是快速定位问题的关键。除了记录INFO级别的操作更要将请求的URL、关键参数、响应状态码和错误信息记录在DEBUG或ERROR级别。建议使用像loguru这样更友好的日志库并配置日志滚动归档方便后期排查历史问题。6. 常见问题排查与解决方案实录在实际对接中你几乎一定会遇到下面这些问题。我把它们和解决方案整理成了表格方便快速查阅。问题现象可能原因排查步骤与解决方案400错误提示type must be in [enabled, disabled, auto]请求体或查询参数中某个字段的值不符合API枚举要求。1. 仔细检查报错接口的官方文档确认type字段允许的值。2. 核对代码中传入的值是否完全一致大小写敏感。3. 使用print或日志输出完整的请求体与文档示例对比。400错误提示this models maximum context length is ... tokens请求内容如消息文本、表格单元格内容过长超过了接口限制。1. 此错误常见于AI相关接口检查发送的文本是否超长。2. 对于普通消息飞书单条文本消息有限制需分段发送。3. 对于多维表格检查单个单元格内容是否过多。401认证失败tenant_access_token无效或已过期。1. 检查app_id和app_secret是否正确。2. 检查客户端代码的token缓存和刷新逻辑是否正常。3. 在开放平台后台检查应用是否已被停用。403无权限应用没有调用该接口的权限。1. 登录开放平台在“权限管理”中查看是否已添加对应权限。2. 添加权限后是否已点击“申请发布”并获管理员批准。3. 部分权限如获取部门下所有用户需要申请更高级别的权限。429请求过于频繁触发了飞书API的速率限制。1. 立即停止当前批量请求。2. 在代码中实现指数退避算法的重试机制。3. 检查业务逻辑优化代码减少不必要的API调用如缓存用户信息。4. 查看响应头是否有Retry-After按其指示的时间等待。消息发送成功但收不到1. 机器人不在该群。2. 发送给了错误的chat_id或open_id。3. 群或用户已屏蔽机器人。1. 确认机器人已加入目标群。2. 使用“获取群列表”API确认使用的chat_id是否正确。3. 尝试向其他群或用户发送以排除屏蔽可能。多维表格操作失败字段值不生效字段值格式与字段类型不匹配。1.这是最高频的错误。在表格页面查看字段的确切类型。2. 对照官方文档检查传入的数据格式-日期毫秒时间戳整数-人员{id: “user_id”, “type”: “User”}对象-单选选项ID字符串-附件[{file_token: “xxx”}]列表事件订阅URL验证不通过1. 服务器未正确响应challenge。2. 网络问题导致飞书无法访问你的URL。3. URL使用了HTTPS但证书有问题。1. 确保你的服务器公网可访问且/webhook路径的GET请求能返回{“challenge”: “xxx”}。2. 使用curl或在线工具手动测试你的URL。3. 本地开发可使用内网穿透工具如ngrok提供临时公网地址。收到事件但解密失败未在代码中实现解密逻辑或加密密钥配置错误。1. 在开放平台“事件订阅”中确认已设置“加密密钥”。2. 在代码中严格按照飞书官方提供的加解密SDK或示例处理encrypt字段。最后再分享一个我自己的体会不要重复造轮子。飞书官方为Python提供了SDK (lark-oapi)它封装了认证、请求、加解密等几乎所有底层细节。对于生产环境尤其是需要处理事件订阅的复杂应用直接使用官方SDK是更稳定、更高效的选择。本文从零开始手写是为了让大家彻底理解其原理但在实际项目中评估后引入官方SDK往往能事半功倍。