跳转至

feishu

快速开始

FeishuClient 是一个异步客户端，统一管理应用凭证、令牌的自动获取与缓存，以及带自动重试的 HTTP 传输。推荐通过 async with 使用，以便自动释放底层连接。消息接口以接收者为第一个参数（其 ID 类型会自动推断），消息内容为第二个参数；client.im.send 还会根据内容的形态自动推断消息类型（msg_type）。

Python
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
import asyncio

from feishu import FeishuClient


async def main():
    async with FeishuClient("cli_xxx", "app_secret") as client:
        await client.im.send("oc_xxx", "hello, world!")


asyncio.run(main())

应用凭证也可以通过环境变量 FEISHU_APP_ID / FEISHU_APP_SECRET 提供，此时可省略构造参数：async with FeishuClient() as client: ...。

命名空间

各业务能力按资源划分到不同命名空间下，每个命名空间提供「裸动词」式的 CRUD 方法（如 create / get / update / delete / list）。常用入口：

命名空间 说明
client.im 即时消息：发送、回复、编辑、撤回、转发消息
client.contact.users / client.contact.departments 通讯录：用户与部门
client.bitable.tables 多维表格：数据表、字段与记录
client.calendar.events 日历：日程与参与人
client.approval.instances 审批：审批实例与任务
client.drive.files 云空间：文件的上传、下载、复制与删除

此外还有 client.docx（新版文档）、client.sheets（电子表格）、client.wiki（知识库）、client.board.whiteboards（画板）、client.vc（视频会议）、client.task（任务）、client.oauth（用户身份 OAuth）、client.cards（卡片构建器）等命名空间。

Python
1
2
3
async with FeishuClient("cli_xxx", "app_secret") as client:
    user = await client.contact.users.get("ou_xxx", user_id_type="open_id")
    records = await client.bitable.records.list("app_token", "tbl_xxx")

接收事件

要接收飞书推送的事件（如「接收消息」），先用 EventDispatcher 按事件类型注册异步处理函数，Webhook 接收器与长连接两种接入方式共用同一套处理函数。

Python
1
2
3
4
5
6
7
8
from feishu.events import EventDispatcher

dispatcher = EventDispatcher()


@dispatcher.on("im.message.receive_v1")
async def on_message(event):
    print(event.event_id)

方式一：Webhook 接收器。 当应用有公网回调地址时，用 create_event_app 生成一个可独立运行的 Starlette 应用，默认带签名新鲜度（防重放）与去重保护，以任意 ASGI 服务器（如 uvicorn）运行即可：

Python
1
2
3
4
from feishu.events import create_event_app

app = create_event_app(dispatcher, encrypt_key="ek_secret")
# uvicorn module:app

方式二：长连接（WebSocket）。 当应用没有公网回调地址时，用 feishu.ws.WsClient 主动与飞书建立一条持久 WebSocket 连接（对标 Slack 的 Socket Mode），事件经该连接推送：

Python
1
2
3
4
5
6
import asyncio

from feishu.ws import WsClient

ws = WsClient("cli_xxx", "app_secret", dispatcher)
asyncio.run(ws.start())

长连接依赖可选的 websockets 包，可通过 pip install open-feishu[ws] 安装。

安装

从 PyPI 安装最新的稳定版本：

Bash
1
pip install open-feishu

如果需要使用 feishu-mcp 命令，请安装 MCP 可选依赖：

Bash
1
pip install "open-feishu[mcp]"

从源代码安装最新版本：

Bash
1
pip install git+https://github.com/ZhiyuanChen/open-feishu.git

许可证

SPDX-License-Identifier: AGPL-3.0-or-later

feishu

feishu.RetryPolicy dataclass

传输层重试策略。

定义对可重试错误（HTTP 5xx 与 429 限流、网络错误）的最大重试次数与退避时间。 退避采用指数增长并以 max_delay 封顶；启用 jitter 时在区间内随机抖动以避免 雷鸣效应。若服务器给出了限流重置提示，则优先使用该提示值（同样以 max_delay 封顶） 作为等待时间。max_elapsed 为整个重试过程的总耗时预算上限，一旦累计退避耗时超过该值 便不再继续重试，避免在持续故障时无限拖延。

参数：

名称 类型 描述 默认

max_attempts

int

最大尝试次数（含首次请求），默认 3。

 3

base_delay

float

退避基准时间（秒），默认 0.5。

 0.5

max_delay

float

单次退避时间上限（秒），默认 30.0。

 30.0

jitter

bool

是否对退避时间施加随机抖动，默认 True。

 True

max_elapsed

float | None

整个重试过程的总退避耗时预算上限（秒）；为 None 时取 max_delay * max_attempts。超过该预算后停止重试。

 None

示例：

Python Console Session
1
2
3
4
5
6
7
8
9
>>> policy = RetryPolicy(base_delay=0.5, jitter=False)
>>> policy.delay(1, None)
0.5
>>> policy.delay(3, None)
2.0
>>> policy.delay(1, reset_after=7.0)
7.0
>>> RetryPolicy(max_delay=5.0, jitter=False).delay(1, reset_after=7.0)
5.0
源代码位于： feishu/_transport.py
Python
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
@dataclass
class RetryPolicy:
    r"""
    传输层重试策略。

    定义对可重试错误（HTTP 5xx 与 429 限流、网络错误）的最大重试次数与退避时间。
    退避采用指数增长并以 `max_delay` 封顶；启用 `jitter` 时在区间内随机抖动以避免
    雷鸣效应。若服务器给出了限流重置提示，则优先使用该提示值（同样以 `max_delay` 封顶）
    作为等待时间。`max_elapsed` 为整个重试过程的总耗时预算上限，一旦累计退避耗时超过该值
    便不再继续重试，避免在持续故障时无限拖延。

    Args:
        max_attempts: 最大尝试次数（含首次请求），默认 3。
        base_delay: 退避基准时间（秒），默认 0.5。
        max_delay: 单次退避时间上限（秒），默认 30.0。
        jitter: 是否对退避时间施加随机抖动，默认 `True`。
        max_elapsed: 整个重试过程的总退避耗时预算上限（秒）；为 `None` 时取
            `max_delay * max_attempts`。超过该预算后停止重试。

    Examples:
        >>> policy = RetryPolicy(base_delay=0.5, jitter=False)
        >>> policy.delay(1, None)
        0.5
        >>> policy.delay(3, None)
        2.0
        >>> policy.delay(1, reset_after=7.0)
        7.0
        >>> RetryPolicy(max_delay=5.0, jitter=False).delay(1, reset_after=7.0)
        5.0
    """

    max_attempts: int = 3
    base_delay: float = 0.5
    max_delay: float = 30.0
    jitter: bool = True
    max_elapsed: float | None = None

    @classmethod
    def default(cls) -> RetryPolicy:
        r"""
        返回使用全部默认参数的重试策略。

        Returns:
            默认配置的 [feishu.RetryPolicy][] 实例。

        Examples:
            >>> RetryPolicy.default().max_attempts
            3
        """
        return cls()

    @property
    def elapsed_budget(self) -> float:
        r"""
        整个重试过程的总退避耗时预算（秒）。

        若显式设置了 `max_elapsed` 则采用之，否则回退到 `max_delay * max_attempts`。

        Returns:
            总退避耗时预算（秒）。

        Examples:
            >>> RetryPolicy(max_delay=30.0, max_attempts=3).elapsed_budget
            90.0
            >>> RetryPolicy(max_elapsed=10.0).elapsed_budget
            10.0
        """
        if self.max_elapsed is not None:
            return self.max_elapsed
        return self.max_delay * self.max_attempts

    def delay(self, attempt: int, reset_after: float | None) -> float:
        r"""
        计算第 `attempt` 次重试前应等待的秒数。

        若提供了 `reset_after`（服务器限流重置提示），则采用该值并以 `max_delay` 封顶，
        防止异常大的服务端提示导致过长等待；否则按 `base_delay * 2 ** (attempt - 1)`
        指数退避并以 `max_delay` 封顶，启用 `jitter` 时再在 `[0, delay]` 区间内随机取值。

        Args:
            attempt: 当前重试序号，从 1 开始。
            reset_after: 服务器给出的限流重置等待秒数；为 `None` 时按指数退避计算。

        Returns:
            建议等待的秒数。

        Examples:
            >>> RetryPolicy(base_delay=0.5, jitter=False).delay(2, None)
            1.0
            >>> RetryPolicy(max_delay=5.0).delay(1, reset_after=7.0)
            5.0
        """
        if reset_after is not None:
            return min(self.max_delay, reset_after)
        delay = min(self.max_delay, self.base_delay * (2 ** (attempt - 1)))
        if self.jitter:
            delay = random.uniform(0, delay) if delay else 0.0
        return delay

elapsed_budget property

Python
elapsed_budget: float

整个重试过程的总退避耗时预算（秒）。

若显式设置了 max_elapsed 则采用之，否则回退到 max_delay * max_attempts。

返回：

类型 描述
float

总退避耗时预算（秒）。

示例：

Python Console Session
1
2
3
4
>>> RetryPolicy(max_delay=30.0, max_attempts=3).elapsed_budget
90.0
>>> RetryPolicy(max_elapsed=10.0).elapsed_budget
10.0

default classmethod

Python
default() -> RetryPolicy

返回使用全部默认参数的重试策略。

返回：

类型 描述
RetryPolicy

默认配置的 feishu.RetryPolicy 实例。

示例：

Python Console Session
1
2
>>> RetryPolicy.default().max_attempts
3
源代码位于： feishu/_transport.py
Python
86
87
88
89
90
91
92
93
94
95
96
97
98
@classmethod
def default(cls) -> RetryPolicy:
    r"""
    返回使用全部默认参数的重试策略。

    Returns:
        默认配置的 [feishu.RetryPolicy][] 实例。

    Examples:
        >>> RetryPolicy.default().max_attempts
        3
    """
    return cls()

delay

Python
delay(attempt: int, reset_after: float | None) -> float

计算第 attempt 次重试前应等待的秒数。

若提供了 reset_after（服务器限流重置提示），则采用该值并以 max_delay 封顶， 防止异常大的服务端提示导致过长等待；否则按 base_delay * 2 ** (attempt - 1) 指数退避并以 max_delay 封顶，启用 jitter 时再在 [0, delay] 区间内随机取值。

参数：

名称 类型 描述 默认

attempt

int

当前重试序号，从 1 开始。

 必需

reset_after

float | None

服务器给出的限流重置等待秒数；为 None 时按指数退避计算。

 必需

返回：

类型 描述
float

建议等待的秒数。

示例：

Python Console Session
1
2
3
4
>>> RetryPolicy(base_delay=0.5, jitter=False).delay(2, None)
1.0
>>> RetryPolicy(max_delay=5.0).delay(1, reset_after=7.0)
5.0
源代码位于： feishu/_transport.py
Python
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def delay(self, attempt: int, reset_after: float | None) -> float:
    r"""
    计算第 `attempt` 次重试前应等待的秒数。

    若提供了 `reset_after`（服务器限流重置提示），则采用该值并以 `max_delay` 封顶，
    防止异常大的服务端提示导致过长等待；否则按 `base_delay * 2 ** (attempt - 1)`
    指数退避并以 `max_delay` 封顶，启用 `jitter` 时再在 `[0, delay]` 区间内随机取值。

    Args:
        attempt: 当前重试序号，从 1 开始。
        reset_after: 服务器给出的限流重置等待秒数；为 `None` 时按指数退避计算。

    Returns:
        建议等待的秒数。

    Examples:
        >>> RetryPolicy(base_delay=0.5, jitter=False).delay(2, None)
        1.0
        >>> RetryPolicy(max_delay=5.0).delay(1, reset_after=7.0)
        5.0
    """
    if reset_after is not None:
        return min(self.max_delay, reset_after)
    delay = min(self.max_delay, self.base_delay * (2 ** (attempt - 1)))
    if self.jitter:
        delay = random.uniform(0, delay) if delay else 0.0
    return delay