跳转至

session

feishu.agent.session

SessionStore

Bases: Protocol

会话历史存储协议，是自定义持久化后端的扩展契约。

feishu.agent.loop.AgentEngine 通过该协议读写各会话的对话历史；内置实现为 feishu.agent.session.InMemorySessionStore，可自行实现该协议接入数据库等持久化后端。该协议标注了 runtime_checkable，可用 isinstance 校验实现是否符合契约。

示例：

Python Console Session
1
2
>>> isinstance(InMemorySessionStore(), SessionStore)
True
源代码位于： feishu/agent/session.py
Python
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@runtime_checkable
class SessionStore(Protocol):
    r"""
    会话历史存储协议，是自定义持久化后端的扩展契约。

    [feishu.agent.loop.AgentEngine][] 通过该协议读写各会话的对话历史；内置实现为
    [feishu.agent.session.InMemorySessionStore][]，可自行实现该协议接入数据库等持久化后端。该协议标注了
    `runtime_checkable`，可用 `isinstance` 校验实现是否符合契约。

    Examples:
        >>> isinstance(InMemorySessionStore(), SessionStore)
        True
    """

    async def get(self, session_id: str) -> list[Message]:
        r"""读取指定会话的全部历史消息。"""
        ...

    async def append(self, session_id: str, *messages: Message) -> None:
        r"""向指定会话追加一条或多条消息。"""
        ...

    async def set(self, session_id: str, messages: list[Message]) -> None:
        r"""以给定的消息列表整体替换指定会话的历史。"""
        ...

    async def clear(self, session_id: str) -> None:
        r"""清空指定会话的历史（彻底删除，而非隐藏）。"""
        ...

    async def updated_at(self, session_id: str) -> float | None:
        r"""返回指定会话最近写入时间戳；未知会话返回 `None`。"""
        ...

get async

Python
get(session_id: str) -> list[Message]

读取指定会话的全部历史消息。

源代码位于： feishu/agent/session.py
Python
50
51
52
async def get(self, session_id: str) -> list[Message]:
    r"""读取指定会话的全部历史消息。"""
    ...

append async

Python
append(session_id: str, *messages: Message) -> None

向指定会话追加一条或多条消息。

源代码位于： feishu/agent/session.py
Python
54
55
56
async def append(self, session_id: str, *messages: Message) -> None:
    r"""向指定会话追加一条或多条消息。"""
    ...

set async

Python
set(session_id: str, messages: list[Message]) -> None

以给定的消息列表整体替换指定会话的历史。

源代码位于： feishu/agent/session.py
Python
58
59
60
async def set(self, session_id: str, messages: list[Message]) -> None:
    r"""以给定的消息列表整体替换指定会话的历史。"""
    ...

clear async

Python
clear(session_id: str) -> None

清空指定会话的历史（彻底删除，而非隐藏）。

源代码位于： feishu/agent/session.py
Python
62
63
64
async def clear(self, session_id: str) -> None:
    r"""清空指定会话的历史（彻底删除，而非隐藏）。"""
    ...

updated_at async

Python
updated_at(session_id: str) -> float | None

返回指定会话最近写入时间戳；未知会话返回 None。

源代码位于： feishu/agent/session.py
Python
66
67
68
async def updated_at(self, session_id: str) -> float | None:
    r"""返回指定会话最近写入时间戳；未知会话返回 `None`。"""
    ...

InMemorySessionStore

基于内存的 feishu.agent.session.SessionStore 实现。

将各会话历史保存在进程内字典中，写操作以锁保护因而并发安全。仅适用于单进程、可接受重启即丢失历史的 场景；生产环境请自行实现 feishu.agent.session.SessionStore 接入持久化后端。

示例：

Python Console Session
1
2
3
4
5
6
7
8
9
>>> import asyncio
>>> from feishu.agent.llm import Message, TextPart
>>> store = InMemorySessionStore()
>>> async def demo():
...     await store.append("oc_1", Message(role="user", content=[TextPart(text="你好")]))
...     history = await store.get("oc_1")
...     return history[0].content[0].text
>>> asyncio.run(demo())
'你好'
源代码位于： feishu/agent/session.py
Python
 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
class InMemorySessionStore:
    r"""
    基于内存的 [feishu.agent.session.SessionStore][] 实现。

    将各会话历史保存在进程内字典中，写操作以锁保护因而并发安全。仅适用于单进程、可接受重启即丢失历史的
    场景；生产环境请自行实现 [feishu.agent.session.SessionStore][] 接入持久化后端。

    Examples:
        >>> import asyncio
        >>> from feishu.agent.llm import Message, TextPart
        >>> store = InMemorySessionStore()
        >>> async def demo():
        ...     await store.append("oc_1", Message(role="user", content=[TextPart(text="你好")]))
        ...     history = await store.get("oc_1")
        ...     return history[0].content[0].text
        >>> asyncio.run(demo())
        '你好'
    """

    def __init__(self) -> None:
        self._store: dict[str, list[Message]] = {}
        self._updated_at: dict[str, float] = {}
        self._lock = asyncio.Lock()

    async def get(self, session_id: str) -> list[Message]:
        r"""
        读取指定会话的全部历史消息。

        Args:
            session_id: 会话标识。

        Returns:
            历史消息的副本；会话不存在时返回空列表。对返回列表的修改不会影响内部状态。
        """
        return list(self._store.get(session_id, []))

    async def append(self, session_id: str, *messages: Message) -> None:
        r"""
        向指定会话追加一条或多条消息。

        Args:
            session_id: 会话标识。
            *messages: 待追加的消息。
        """
        async with self._lock:
            bucket = self._store.get(session_id)
            if bucket is None:  # double-checked read after acquiring the lock
                bucket = self._store.setdefault(session_id, [])
            bucket.extend(messages)
            self._updated_at[session_id] = time.time()

    async def set(self, session_id: str, messages: list[Message]) -> None:
        r"""
        以给定的消息列表整体替换指定会话的历史。

        Args:
            session_id: 会话标识。
            messages: 用于替换的消息列表，将被拷贝保存。
        """
        async with self._lock:
            self._store[session_id] = list(messages)
            self._updated_at[session_id] = time.time()

    async def clear(self, session_id: str) -> None:
        r"""清空指定会话的历史（彻底删除该会话条目）。"""
        async with self._lock:
            self._store.pop(session_id, None)
            self._updated_at.pop(session_id, None)

    async def updated_at(self, session_id: str) -> float | None:
        r"""返回指定会话最近写入时间戳；未知会话返回 `None`。"""
        return self._updated_at.get(session_id)

get async

Python
get(session_id: str) -> list[Message]

读取指定会话的全部历史消息。

参数：

名称 类型 描述 默认
session_id
str

会话标识。

 必需

返回：

类型 描述
list[Message]

历史消息的副本；会话不存在时返回空列表。对返回列表的修改不会影响内部状态。
源代码位于： feishu/agent/session.py
Python
 95
 96
 97
 98
 99
100
101
102
103
104
105
async def get(self, session_id: str) -> list[Message]:
    r"""
    读取指定会话的全部历史消息。

    Args:
        session_id: 会话标识。

    Returns:
        历史消息的副本；会话不存在时返回空列表。对返回列表的修改不会影响内部状态。
    """
    return list(self._store.get(session_id, []))

append async

Python
append(session_id: str, *messages: Message) -> None

向指定会话追加一条或多条消息。

参数：

名称 类型 描述 默认
session_id
str

会话标识。

 必需
*messages
Message

待追加的消息。

 ()
源代码位于： feishu/agent/session.py
Python
107
108
109
110
111
112
113
114
115
116
117
118
119
120
async def append(self, session_id: str, *messages: Message) -> None:
    r"""
    向指定会话追加一条或多条消息。

    Args:
        session_id: 会话标识。
        *messages: 待追加的消息。
    """
    async with self._lock:
        bucket = self._store.get(session_id)
        if bucket is None:  # double-checked read after acquiring the lock
            bucket = self._store.setdefault(session_id, [])
        bucket.extend(messages)
        self._updated_at[session_id] = time.time()

set async

Python
set(session_id: str, messages: list[Message]) -> None

以给定的消息列表整体替换指定会话的历史。

参数：

名称 类型 描述 默认
session_id
str

会话标识。

 必需
messages
list[Message]

用于替换的消息列表，将被拷贝保存。

 必需
源代码位于： feishu/agent/session.py
Python
122
123
124
125
126
127
128
129
130
131
132
async def set(self, session_id: str, messages: list[Message]) -> None:
    r"""
    以给定的消息列表整体替换指定会话的历史。

    Args:
        session_id: 会话标识。
        messages: 用于替换的消息列表，将被拷贝保存。
    """
    async with self._lock:
        self._store[session_id] = list(messages)
        self._updated_at[session_id] = time.time()

clear async

Python
clear(session_id: str) -> None

清空指定会话的历史（彻底删除该会话条目）。

源代码位于： feishu/agent/session.py
Python
134
135
136
137
138
async def clear(self, session_id: str) -> None:
    r"""清空指定会话的历史（彻底删除该会话条目）。"""
    async with self._lock:
        self._store.pop(session_id, None)
        self._updated_at.pop(session_id, None)

updated_at async

Python
updated_at(session_id: str) -> float | None

返回指定会话最近写入时间戳；未知会话返回 None。

源代码位于： feishu/agent/session.py
Python
140
141
142
async def updated_at(self, session_id: str) -> float | None:
    r"""返回指定会话最近写入时间戳；未知会话返回 `None`。"""
    return self._updated_at.get(session_id)

PendingApproval dataclass

一次挂起的工具审批，记录恢复对话所需的全部上下文。

当工具的 requires_approval 为 True 时，feishu.agent.loop.AgentEngine 会创建该记录并发送审批卡片； 用户在卡片上批准或拒绝后，依据其中保存的会话与工具调用信息恢复本轮对话。

示例：

Python Console Session
1
2
3
4
5
6
7
8
9
>>> approval = PendingApproval(
...     approval_id="ap_1",
...     session_id="oc_1",
...     tool_call_id="c1",
...     tool_name="deploy",
...     arguments={"env": "prod"},
... )
>>> approval.tool_name
'deploy'
源代码位于： feishu/agent/session.py
Python
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@dataclass
class PendingApproval:
    r"""
    一次挂起的工具审批，记录恢复对话所需的全部上下文。

    当工具的 `requires_approval` 为 `True` 时，[feishu.agent.loop.AgentEngine][] 会创建该记录并发送审批卡片；
    用户在卡片上批准或拒绝后，依据其中保存的会话与工具调用信息恢复本轮对话。

    Examples:
        >>> approval = PendingApproval(
        ...     approval_id="ap_1",
        ...     session_id="oc_1",
        ...     tool_call_id="c1",
        ...     tool_name="deploy",
        ...     arguments={"env": "prod"},
        ... )
        >>> approval.tool_name
        'deploy'
    """

    approval_id: str
    session_id: str
    tool_call_id: str
    tool_name: str
    arguments: dict[str, Any]
    # Optional integrity / idempotency / ownership metadata. All default so that
    # a minimal PendingApproval(approval_id, session_id, tool_call_id, tool_name,
    # arguments) keeps working; durable, tamper-checked stores populate the rest.
    payload_sha256: str | None = None
    idempotency_key: str | None = None
    owner_user_keys: tuple[str, ...] = ()
    tenant_key: str | None = None
    chat_id: str | None = None
    created_message_id: str | None = None
    created_event_id: str | None = None
    created_at: int | None = None
    state: str = "awaiting_confirmation"
    extra: dict[str, Any] = field(default_factory=dict)

PendingAuthorization dataclass

一次挂起的用户授权，记录 OAuth callback 后恢复原工具调用所需的上下文。

当工具返回 NEEDS_USER_AUTH 时，feishu.agent.loop.AgentEngine 会创建该记录、发送授权卡片并挂起本轮； OAuth callback 完成并保存用户 token 后，产品调用 Agent.resume_authorization，依据其中保存的工具调用恢复对话。

源代码位于： feishu/agent/session.py
Python
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@dataclass
class PendingAuthorization:
    r"""
    一次挂起的用户授权，记录 OAuth callback 后恢复原工具调用所需的上下文。

    当工具返回 `NEEDS_USER_AUTH` 时，[feishu.agent.loop.AgentEngine][] 会创建该记录、发送授权卡片并挂起本轮；
    OAuth callback 完成并保存用户 token 后，产品调用 `Agent.resume_authorization`，依据其中保存的工具调用恢复对话。
    """

    authorization_id: str
    session_id: str
    tool_call_id: str
    tool_name: str
    arguments: dict[str, Any]
    scopes: tuple[str, ...] = ()
    owner_user_keys: tuple[str, ...] = ()
    tenant_key: str | None = None
    chat_id: str | None = None
    created_message_id: str | None = None
    created_event_id: str | None = None
    created_at: int | None = None
    state: str = "awaiting_authorization"
    extra: dict[str, Any] = field(default_factory=dict)

ClaimResult

Bases: str, Enum

一次审批认领（claim）的结果，是审批执行前的并发安全闸门。

feishu.agent.approval.ApprovalEngine 在执行工具前先 claim 对应审批，依据返回值决定放行或拒绝： 仅 CLAIMED 允许继续执行，其余值各自对应一种不可执行的情形。由于继承自 str，枚举成员可直接与字符串 字面量比较。

示例：

Python Console Session
1
2
3
4
>>> ClaimResult.CLAIMED == "claimed"
True
>>> ClaimResult("tampered") is ClaimResult.TAMPERED
True
源代码位于： feishu/agent/session.py
Python
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
class ClaimResult(str, Enum):
    r"""
    一次审批认领（claim）的结果，是审批执行前的并发安全闸门。

    [feishu.agent.approval.ApprovalEngine][] 在执行工具前先 `claim` 对应审批，依据返回值决定放行或拒绝：
    仅 `CLAIMED` 允许继续执行，其余值各自对应一种不可执行的情形。由于继承自 `str`，枚举成员可直接与字符串
    字面量比较。

    Examples:
        >>> ClaimResult.CLAIMED == "claimed"
        True
        >>> ClaimResult("tampered") is ClaimResult.TAMPERED
        True
    """

    CLAIMED = "claimed"  # state flipped awaiting_confirmation -> executing; proceed
    ALREADY_CLAIMED = "already_claimed"  # another confirm already claimed/executed it
    SUPERSEDED = "superseded"  # a newer proposal for the same operation replaced it
    TAMPERED = "tampered"  # expected_payload_sha256 != stored payload_sha256
    EXPIRED = "expired"  # TTL elapsed before confirmation
    MISSING = "missing"  # no such approval (unknown id or already resolved)

PendingApprovalStore

Bases: Protocol

挂起审批存储协议，是自定义审批持久化后端的扩展契约。

feishu.agent.loop.AgentEngine 通过该协议保存与取回挂起的 feishu.agent.session.PendingApproval； 内置实现为 feishu.agent.session.InMemoryPendingApprovalStore。该协议标注了 runtime_checkable， 可用 isinstance 校验实现是否符合契约。

示例：

Python Console Session
1
2
>>> isinstance(InMemoryPendingApprovalStore(), PendingApprovalStore)
True
源代码位于： feishu/agent/session.py
Python
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
@runtime_checkable
class PendingApprovalStore(Protocol):
    r"""
    挂起审批存储协议，是自定义审批持久化后端的扩展契约。

    [feishu.agent.loop.AgentEngine][] 通过该协议保存与取回挂起的 [feishu.agent.session.PendingApproval][]；
    内置实现为 [feishu.agent.session.InMemoryPendingApprovalStore][]。该协议标注了 `runtime_checkable`，
    可用 `isinstance` 校验实现是否符合契约。

    Examples:
        >>> isinstance(InMemoryPendingApprovalStore(), PendingApprovalStore)
        True
    """

    async def put(self, approval: PendingApproval) -> None:
        r"""保存一次挂起的审批。"""
        ...

    async def pop(self, approval_id: str) -> PendingApproval | None:
        r"""按 `approval_id` 取出并移除一次挂起的审批，不存在时返回 `None`。"""
        ...

    async def get(self, approval_id: str) -> PendingApproval | None:
        r"""按 `approval_id` 读取挂起的审批而不移除，不存在时返回 `None`。"""
        ...

    async def claim(self, approval_id: str, *, expected_payload_sha256: str | None = None) -> ClaimResult:
        r"""
        原子地认领一次审批（`awaiting_confirmation` -> `executing`），返回 [feishu.agent.session.ClaimResult][]。

        这是防重复执行与防篡改的并发闸门：提供 `expected_payload_sha256` 时须与存储的负载摘要一致，否则返回
        `TAMPERED`；已被认领/执行返回 `ALREADY_CLAIMED`；不存在返回 `MISSING`。仅 `CLAIMED` 允许继续执行。
        """
        ...

    async def complete(self, approval_id: str, *, outcome: str) -> None:
        r"""标记一次审批的最终处置：成功/拒绝/取消即移除，结果未知则冻结以防重复执行。"""
        ...

    async def update(self, approval_id: str, mutator: Callable[[PendingApproval], tuple[T, PendingApproval]]) -> T:
        r"""以 compare-and-swap 方式原子更新一次审批：`mutator(旧值)` 返回 `(返回值, 新值)`。"""
        ...

put async

Python
put(approval: PendingApproval) -> None

保存一次挂起的审批。

源代码位于： feishu/agent/session.py
Python
247
248
249
async def put(self, approval: PendingApproval) -> None:
    r"""保存一次挂起的审批。"""
    ...

pop async

Python
pop(approval_id: str) -> PendingApproval | None

按 approval_id 取出并移除一次挂起的审批，不存在时返回 None。

源代码位于： feishu/agent/session.py
Python
251
252
253
async def pop(self, approval_id: str) -> PendingApproval | None:
    r"""按 `approval_id` 取出并移除一次挂起的审批，不存在时返回 `None`。"""
    ...

get async

Python
get(approval_id: str) -> PendingApproval | None

按 approval_id 读取挂起的审批而不移除，不存在时返回 None。

源代码位于： feishu/agent/session.py
Python
255
256
257
async def get(self, approval_id: str) -> PendingApproval | None:
    r"""按 `approval_id` 读取挂起的审批而不移除，不存在时返回 `None`。"""
    ...

claim async

Python
claim(approval_id: str, *, expected_payload_sha256: str | None = None) -> ClaimResult

原子地认领一次审批（awaiting_confirmation -> executing），返回 feishu.agent.session.ClaimResult。

这是防重复执行与防篡改的并发闸门：提供 expected_payload_sha256 时须与存储的负载摘要一致，否则返回 TAMPERED；已被认领/执行返回 ALREADY_CLAIMED；不存在返回 MISSING。仅 CLAIMED 允许继续执行。

源代码位于： feishu/agent/session.py
Python
259
260
261
262
263
264
265
266
async def claim(self, approval_id: str, *, expected_payload_sha256: str | None = None) -> ClaimResult:
    r"""
    原子地认领一次审批（`awaiting_confirmation` -> `executing`），返回 [feishu.agent.session.ClaimResult][]。

    这是防重复执行与防篡改的并发闸门：提供 `expected_payload_sha256` 时须与存储的负载摘要一致，否则返回
    `TAMPERED`；已被认领/执行返回 `ALREADY_CLAIMED`；不存在返回 `MISSING`。仅 `CLAIMED` 允许继续执行。
    """
    ...

complete async

Python
complete(approval_id: str, *, outcome: str) -> None

标记一次审批的最终处置：成功/拒绝/取消即移除，结果未知则冻结以防重复执行。

源代码位于： feishu/agent/session.py
Python
268
269
270
async def complete(self, approval_id: str, *, outcome: str) -> None:
    r"""标记一次审批的最终处置：成功/拒绝/取消即移除，结果未知则冻结以防重复执行。"""
    ...

update async

Python
update(approval_id: str, mutator: Callable[[PendingApproval], tuple[T, PendingApproval]]) -> T

以 compare-and-swap 方式原子更新一次审批：mutator(旧值) 返回 (返回值, 新值)。

源代码位于： feishu/agent/session.py
Python
272
273
274
async def update(self, approval_id: str, mutator: Callable[[PendingApproval], tuple[T, PendingApproval]]) -> T:
    r"""以 compare-and-swap 方式原子更新一次审批：`mutator(旧值)` 返回 `(返回值, 新值)`。"""
    ...

InMemoryPendingApprovalStore

基于内存的 feishu.agent.session.PendingApprovalStore 实现。

将挂起审批保存在进程内字典中，写操作以锁保护因而并发安全。每个审批仅可被取出一次，取出即移除，可天然 防止重复执行。仅适用于单进程场景；生产环境请自行实现 feishu.agent.session.PendingApprovalStore。

示例：

Python Console Session
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
>>> import asyncio
>>> store = InMemoryPendingApprovalStore()
>>> approval = PendingApproval(
...     approval_id="ap_1",
...     session_id="oc_1",
...     tool_call_id="c1",
...     tool_name="deploy",
...     arguments={"env": "prod"},
... )
>>> async def demo():
...     await store.put(approval)
...     first = await store.pop("ap_1")
...     second = await store.pop("ap_1")
...     return first.tool_name, second
>>> asyncio.run(demo())
('deploy', None)
源代码位于： feishu/agent/session.py
Python
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
class InMemoryPendingApprovalStore:
    r"""
    基于内存的 [feishu.agent.session.PendingApprovalStore][] 实现。

    将挂起审批保存在进程内字典中，写操作以锁保护因而并发安全。每个审批仅可被取出一次，取出即移除，可天然
    防止重复执行。仅适用于单进程场景；生产环境请自行实现 [feishu.agent.session.PendingApprovalStore][]。

    Examples:
        >>> import asyncio
        >>> store = InMemoryPendingApprovalStore()
        >>> approval = PendingApproval(
        ...     approval_id="ap_1",
        ...     session_id="oc_1",
        ...     tool_call_id="c1",
        ...     tool_name="deploy",
        ...     arguments={"env": "prod"},
        ... )
        >>> async def demo():
        ...     await store.put(approval)
        ...     first = await store.pop("ap_1")
        ...     second = await store.pop("ap_1")
        ...     return first.tool_name, second
        >>> asyncio.run(demo())
        ('deploy', None)
    """

    def __init__(self) -> None:
        self._store: dict[str, PendingApproval] = {}
        self._lock = asyncio.Lock()

    async def put(self, approval: PendingApproval) -> None:
        r"""
        保存一次挂起的审批。

        Args:
            approval: 待保存的 [feishu.agent.session.PendingApproval][]。
        """
        async with self._lock:
            self._store[approval.approval_id] = approval

    async def pop(self, approval_id: str) -> PendingApproval | None:
        r"""
        按 `approval_id` 取出并移除一次挂起的审批。

        Args:
            approval_id: 审批标识。

        Returns:
            对应的 [feishu.agent.session.PendingApproval][]；不存在或已被取出时返回 `None`。
        """
        async with self._lock:
            return self._store.pop(approval_id, None)

    async def get(self, approval_id: str) -> PendingApproval | None:
        r"""
        按 `approval_id` 读取挂起的审批而不移除。

        Args:
            approval_id: 审批标识。

        Returns:
            对应的 [feishu.agent.session.PendingApproval][]；不存在时返回 `None`。
        """
        return self._store.get(approval_id)

    async def claim(self, approval_id: str, *, expected_payload_sha256: str | None = None) -> ClaimResult:
        r"""
        原子地认领一次审批，返回 [feishu.agent.session.ClaimResult][]。

        在锁内完成「存在性 + 篡改 + 状态」三项校验，并在通过时将状态翻转为 `executing`，从而保证同一审批不会
        被并发确认重复执行。

        Args:
            approval_id: 审批标识。
            expected_payload_sha256: 卡片回传携带的负载摘要；提供时须与存储值一致，否则返回 `TAMPERED`。

        Returns:
            认领结果；仅 `CLAIMED` 表示可继续执行。

        Examples:
            >>> import asyncio
            >>> store = InMemoryPendingApprovalStore()
            >>> approval = PendingApproval(
            ...     approval_id="ap_1", session_id="oc_1", tool_call_id="c1",
            ...     tool_name="deploy", arguments={"env": "prod"}, payload_sha256="abc",
            ... )
            >>> async def demo():
            ...     await store.put(approval)
            ...     bad = await store.claim("ap_1", expected_payload_sha256="zzz")
            ...     ok = await store.claim("ap_1", expected_payload_sha256="abc")
            ...     again = await store.claim("ap_1", expected_payload_sha256="abc")
            ...     return bad.value, ok.value, again.value
            >>> asyncio.run(demo())
            ('tampered', 'claimed', 'already_claimed')
        """
        async with self._lock:
            approval = self._store.get(approval_id)
            if approval is None:
                return ClaimResult.MISSING
            # Fail closed on tampering: when EITHER side has a payload hash, both must match. A stored hash with a
            # missing/None callback hash is a mismatch (so a callback that omits payload_sha256 can't skip the
            # check); a callback hash with no stored hash is also a mismatch. Only when neither exists is there
            # nothing to verify.
            stored_sha = approval.payload_sha256 or ""
            if (stored_sha or expected_payload_sha256 is not None) and expected_payload_sha256 != stored_sha:
                return ClaimResult.TAMPERED
            if approval.state != "awaiting_confirmation":
                return ClaimResult.ALREADY_CLAIMED
            approval.state = "executing"
            return ClaimResult.CLAIMED

    async def complete(self, approval_id: str, *, outcome: str) -> None:
        r"""
        标记一次审批的最终处置。

        成功、拒绝或取消（`executed`/`replayed`/`rejected`/`cancelled`）即移除记录；`retry` 还原为
        `awaiting_confirmation` 以便重试；执行结果未知（`unknown`/`frozen`）则冻结为 `execution_unknown`。

        Args:
            approval_id: 审批标识。
            outcome: 最终处置标签。
        """
        async with self._lock:
            if outcome == "retry":
                approval = self._store.get(approval_id)
                if approval is not None:
                    approval.state = "awaiting_confirmation"
                return
            if outcome in ("unknown", "frozen"):
                approval = self._store.get(approval_id)
                if approval is not None:
                    approval.state = "execution_unknown"
                return
            self._store.pop(approval_id, None)

    async def update(self, approval_id: str, mutator: Callable[[PendingApproval], tuple[T, PendingApproval]]) -> T:
        r"""
        以 compare-and-swap 方式原子更新一次审批。

        在锁内调用 `mutator(旧值)`，其须返回 `(返回值, 新值)`；新值写回存储，返回值回传调用方。

        Args:
            approval_id: 审批标识。
            mutator: 接受旧审批、返回 `(返回值, 新审批)` 的纯函数。

        Returns:
            `mutator` 的返回值。

        Raises:
            KeyError: 审批不存在时抛出。
        """
        async with self._lock:
            approval = self._store.get(approval_id)
            if approval is None:
                raise KeyError(approval_id)
            value, updated = mutator(approval)
            self._store[approval_id] = updated
            return value

put async

Python
put(approval: PendingApproval) -> None

保存一次挂起的审批。

参数：

名称 类型 描述 默认
approval
PendingApproval

待保存的 feishu.agent.session.PendingApproval。

 必需
源代码位于： feishu/agent/session.py
Python
307
308
309
310
311
312
313
314
315
async def put(self, approval: PendingApproval) -> None:
    r"""
    保存一次挂起的审批。

    Args:
        approval: 待保存的 [feishu.agent.session.PendingApproval][]。
    """
    async with self._lock:
        self._store[approval.approval_id] = approval

pop async

Python
pop(approval_id: str) -> PendingApproval | None

按 approval_id 取出并移除一次挂起的审批。

参数：

名称 类型 描述 默认
approval_id
str

审批标识。

 必需

返回：

类型 描述
PendingApproval | None

对应的 feishu.agent.session.PendingApproval；不存在或已被取出时返回 None。
源代码位于： feishu/agent/session.py
Python
317
318
319
320
321
322
323
324
325
326
327
328
async def pop(self, approval_id: str) -> PendingApproval | None:
    r"""
    按 `approval_id` 取出并移除一次挂起的审批。

    Args:
        approval_id: 审批标识。

    Returns:
        对应的 [feishu.agent.session.PendingApproval][]；不存在或已被取出时返回 `None`。
    """
    async with self._lock:
        return self._store.pop(approval_id, None)

get async

Python
get(approval_id: str) -> PendingApproval | None

按 approval_id 读取挂起的审批而不移除。

参数：

名称 类型 描述 默认
approval_id
str

审批标识。

 必需

返回：

类型 描述
PendingApproval | None

对应的 feishu.agent.session.PendingApproval；不存在时返回 None。
源代码位于： feishu/agent/session.py
Python
330
331
332
333
334
335
336
337
338
339
340
async def get(self, approval_id: str) -> PendingApproval | None:
    r"""
    按 `approval_id` 读取挂起的审批而不移除。

    Args:
        approval_id: 审批标识。

    Returns:
        对应的 [feishu.agent.session.PendingApproval][]；不存在时返回 `None`。
    """
    return self._store.get(approval_id)

claim async

Python
claim(approval_id: str, *, expected_payload_sha256: str | None = None) -> ClaimResult

原子地认领一次审批，返回 feishu.agent.session.ClaimResult。

在锁内完成「存在性 + 篡改 + 状态」三项校验，并在通过时将状态翻转为 executing，从而保证同一审批不会 被并发确认重复执行。

参数：

名称 类型 描述 默认
approval_id
str

审批标识。

 必需
expected_payload_sha256
str | None

卡片回传携带的负载摘要；提供时须与存储值一致，否则返回 TAMPERED。

 None

返回：

类型 描述
ClaimResult

认领结果；仅 CLAIMED 表示可继续执行。

示例：

Python Console Session
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
>>> import asyncio
>>> store = InMemoryPendingApprovalStore()
>>> approval = PendingApproval(
...     approval_id="ap_1", session_id="oc_1", tool_call_id="c1",
...     tool_name="deploy", arguments={"env": "prod"}, payload_sha256="abc",
... )
>>> async def demo():
...     await store.put(approval)
...     bad = await store.claim("ap_1", expected_payload_sha256="zzz")
...     ok = await store.claim("ap_1", expected_payload_sha256="abc")
...     again = await store.claim("ap_1", expected_payload_sha256="abc")
...     return bad.value, ok.value, again.value
>>> asyncio.run(demo())
('tampered', 'claimed', 'already_claimed')
源代码位于： feishu/agent/session.py
Python
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
async def claim(self, approval_id: str, *, expected_payload_sha256: str | None = None) -> ClaimResult:
    r"""
    原子地认领一次审批，返回 [feishu.agent.session.ClaimResult][]。

    在锁内完成「存在性 + 篡改 + 状态」三项校验，并在通过时将状态翻转为 `executing`，从而保证同一审批不会
    被并发确认重复执行。

    Args:
        approval_id: 审批标识。
        expected_payload_sha256: 卡片回传携带的负载摘要；提供时须与存储值一致，否则返回 `TAMPERED`。

    Returns:
        认领结果；仅 `CLAIMED` 表示可继续执行。

    Examples:
        >>> import asyncio
        >>> store = InMemoryPendingApprovalStore()
        >>> approval = PendingApproval(
        ...     approval_id="ap_1", session_id="oc_1", tool_call_id="c1",
        ...     tool_name="deploy", arguments={"env": "prod"}, payload_sha256="abc",
        ... )
        >>> async def demo():
        ...     await store.put(approval)
        ...     bad = await store.claim("ap_1", expected_payload_sha256="zzz")
        ...     ok = await store.claim("ap_1", expected_payload_sha256="abc")
        ...     again = await store.claim("ap_1", expected_payload_sha256="abc")
        ...     return bad.value, ok.value, again.value
        >>> asyncio.run(demo())
        ('tampered', 'claimed', 'already_claimed')
    """
    async with self._lock:
        approval = self._store.get(approval_id)
        if approval is None:
            return ClaimResult.MISSING
        # Fail closed on tampering: when EITHER side has a payload hash, both must match. A stored hash with a
        # missing/None callback hash is a mismatch (so a callback that omits payload_sha256 can't skip the
        # check); a callback hash with no stored hash is also a mismatch. Only when neither exists is there
        # nothing to verify.
        stored_sha = approval.payload_sha256 or ""
        if (stored_sha or expected_payload_sha256 is not None) and expected_payload_sha256 != stored_sha:
            return ClaimResult.TAMPERED
        if approval.state != "awaiting_confirmation":
            return ClaimResult.ALREADY_CLAIMED
        approval.state = "executing"
        return ClaimResult.CLAIMED

complete async

Python
complete(approval_id: str, *, outcome: str) -> None

标记一次审批的最终处置。

成功、拒绝或取消（executed/replayed/rejected/cancelled）即移除记录；retry 还原为 awaiting_confirmation 以便重试；执行结果未知（unknown/frozen）则冻结为 execution_unknown。

参数：

名称 类型 描述 默认
approval_id
str

审批标识。

 必需
outcome
str

最终处置标签。

 必需
源代码位于： feishu/agent/session.py
Python
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
async def complete(self, approval_id: str, *, outcome: str) -> None:
    r"""
    标记一次审批的最终处置。

    成功、拒绝或取消（`executed`/`replayed`/`rejected`/`cancelled`）即移除记录；`retry` 还原为
    `awaiting_confirmation` 以便重试；执行结果未知（`unknown`/`frozen`）则冻结为 `execution_unknown`。

    Args:
        approval_id: 审批标识。
        outcome: 最终处置标签。
    """
    async with self._lock:
        if outcome == "retry":
            approval = self._store.get(approval_id)
            if approval is not None:
                approval.state = "awaiting_confirmation"
            return
        if outcome in ("unknown", "frozen"):
            approval = self._store.get(approval_id)
            if approval is not None:
                approval.state = "execution_unknown"
            return
        self._store.pop(approval_id, None)

update async

Python
update(approval_id: str, mutator: Callable[[PendingApproval], tuple[T, PendingApproval]]) -> T

以 compare-and-swap 方式原子更新一次审批。

在锁内调用 mutator(旧值)，其须返回 (返回值, 新值)；新值写回存储，返回值回传调用方。

参数：

名称 类型 描述 默认
approval_id
str

审批标识。

 必需
mutator
Callable[[PendingApproval], tuple[T, PendingApproval]]

接受旧审批、返回 (返回值, 新审批) 的纯函数。

 必需

返回：

类型 描述
T

mutator 的返回值。

引发：

类型 描述
KeyError

审批不存在时抛出。
源代码位于： feishu/agent/session.py
Python
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
async def update(self, approval_id: str, mutator: Callable[[PendingApproval], tuple[T, PendingApproval]]) -> T:
    r"""
    以 compare-and-swap 方式原子更新一次审批。

    在锁内调用 `mutator(旧值)`，其须返回 `(返回值, 新值)`；新值写回存储，返回值回传调用方。

    Args:
        approval_id: 审批标识。
        mutator: 接受旧审批、返回 `(返回值, 新审批)` 的纯函数。

    Returns:
        `mutator` 的返回值。

    Raises:
        KeyError: 审批不存在时抛出。
    """
    async with self._lock:
        approval = self._store.get(approval_id)
        if approval is None:
            raise KeyError(approval_id)
        value, updated = mutator(approval)
        self._store[approval_id] = updated
        return value

PendingAuthorizationStore

Bases: Protocol

挂起授权存储协议，是 OAuth 授权后自动恢复工具调用的扩展契约。

feishu.agent.loop.AgentEngine 通过该协议保存与取回挂起的 feishu.agent.session.PendingAuthorization； 内置实现为 feishu.agent.session.InMemoryPendingAuthorizationStore。

源代码位于： feishu/agent/session.py
Python
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
@runtime_checkable
class PendingAuthorizationStore(Protocol):
    r"""
    挂起授权存储协议，是 OAuth 授权后自动恢复工具调用的扩展契约。

    [feishu.agent.loop.AgentEngine][] 通过该协议保存与取回挂起的 [feishu.agent.session.PendingAuthorization][]；
    内置实现为 [feishu.agent.session.InMemoryPendingAuthorizationStore][]。
    """

    async def put(self, authorization: PendingAuthorization) -> None:
        r"""保存一次挂起授权。"""
        ...

    async def get(self, authorization_id: str) -> PendingAuthorization | None:
        r"""按 `authorization_id` 读取挂起授权而不移除，不存在时返回 `None`。"""
        ...

    async def pop(self, authorization_id: str) -> PendingAuthorization | None:
        r"""按 `authorization_id` 取出并移除一次挂起授权，不存在时返回 `None`。"""
        ...

    async def claim(self, authorization_id: str) -> ClaimResult:
        r"""原子地认领一次授权（`awaiting_authorization` -> `executing`），仅 `CLAIMED` 允许恢复工具。"""
        ...

    async def complete(self, authorization_id: str, *, outcome: str) -> None:
        r"""
        标记一次授权恢复的最终处置。

        `retry` 还原为 `awaiting_authorization`；`unknown`/`frozen` 冻结为 `execution_unknown`；其余终态
        （如 `executed`/`failed`/`cancelled`/`expired`）移除记录。
        """
        ...

    async def update(
        self,
        authorization_id: str,
        mutator: Callable[[PendingAuthorization], tuple[T, PendingAuthorization]],
    ) -> T:
        r"""以 compare-and-swap 方式原子更新一次授权：`mutator(旧值)` 返回 `(返回值, 新值)`。"""

put async

Python
put(authorization: PendingAuthorization) -> None

保存一次挂起授权。

源代码位于： feishu/agent/session.py
Python
446
447
448
async def put(self, authorization: PendingAuthorization) -> None:
    r"""保存一次挂起授权。"""
    ...

get async

Python
get(authorization_id: str) -> PendingAuthorization | None

按 authorization_id 读取挂起授权而不移除，不存在时返回 None。

源代码位于： feishu/agent/session.py
Python
450
451
452
async def get(self, authorization_id: str) -> PendingAuthorization | None:
    r"""按 `authorization_id` 读取挂起授权而不移除，不存在时返回 `None`。"""
    ...

pop async

Python
pop(authorization_id: str) -> PendingAuthorization | None

按 authorization_id 取出并移除一次挂起授权，不存在时返回 None。

源代码位于： feishu/agent/session.py
Python
454
455
456
async def pop(self, authorization_id: str) -> PendingAuthorization | None:
    r"""按 `authorization_id` 取出并移除一次挂起授权，不存在时返回 `None`。"""
    ...

claim async

Python
claim(authorization_id: str) -> ClaimResult

原子地认领一次授权（awaiting_authorization -> executing），仅 CLAIMED 允许恢复工具。

源代码位于： feishu/agent/session.py
Python
458
459
460
async def claim(self, authorization_id: str) -> ClaimResult:
    r"""原子地认领一次授权（`awaiting_authorization` -> `executing`），仅 `CLAIMED` 允许恢复工具。"""
    ...

complete async

Python
complete(authorization_id: str, *, outcome: str) -> None

标记一次授权恢复的最终处置。

retry 还原为 awaiting_authorization；unknown/frozen 冻结为 execution_unknown；其余终态 （如 executed/failed/cancelled/expired）移除记录。

源代码位于： feishu/agent/session.py
Python
462
463
464
465
466
467
468
469
async def complete(self, authorization_id: str, *, outcome: str) -> None:
    r"""
    标记一次授权恢复的最终处置。

    `retry` 还原为 `awaiting_authorization`；`unknown`/`frozen` 冻结为 `execution_unknown`；其余终态
    （如 `executed`/`failed`/`cancelled`/`expired`）移除记录。
    """
    ...

update async

Python
update(authorization_id: str, mutator: Callable[[PendingAuthorization], tuple[T, PendingAuthorization]]) -> T

以 compare-and-swap 方式原子更新一次授权：mutator(旧值) 返回 (返回值, 新值)。

源代码位于： feishu/agent/session.py
Python
471
472
473
474
475
476
async def update(
    self,
    authorization_id: str,
    mutator: Callable[[PendingAuthorization], tuple[T, PendingAuthorization]],
) -> T:
    r"""以 compare-and-swap 方式原子更新一次授权：`mutator(旧值)` 返回 `(返回值, 新值)`。"""

InMemoryPendingAuthorizationStore

基于内存的 feishu.agent.session.PendingAuthorizationStore 实现。

源代码位于： feishu/agent/session.py
Python
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
class InMemoryPendingAuthorizationStore:
    r"""基于内存的 [feishu.agent.session.PendingAuthorizationStore][] 实现。"""

    def __init__(self) -> None:
        self._store: dict[str, PendingAuthorization] = {}
        self._lock = asyncio.Lock()

    async def put(self, authorization: PendingAuthorization) -> None:
        r"""保存一次挂起授权。"""
        async with self._lock:
            self._store[authorization.authorization_id] = authorization

    async def get(self, authorization_id: str) -> PendingAuthorization | None:
        r"""按 `authorization_id` 读取挂起授权而不移除。"""
        async with self._lock:
            return self._store.get(authorization_id)

    async def pop(self, authorization_id: str) -> PendingAuthorization | None:
        r"""按 `authorization_id` 取出并移除一次挂起授权。"""
        async with self._lock:
            return self._store.pop(authorization_id, None)

    async def claim(self, authorization_id: str) -> ClaimResult:
        r"""原子地认领一次授权，返回 [feishu.agent.session.ClaimResult][]。"""
        async with self._lock:
            authorization = self._store.get(authorization_id)
            if authorization is None:
                return ClaimResult.MISSING
            if authorization.state != "awaiting_authorization":
                return ClaimResult.ALREADY_CLAIMED
            authorization.state = "executing"
            return ClaimResult.CLAIMED

    async def complete(self, authorization_id: str, *, outcome: str) -> None:
        r"""标记授权恢复的最终处置；`retry` 重开，`unknown`/`frozen` 冻结，其余终态移除。"""
        async with self._lock:
            if outcome in ("unknown", "frozen"):
                authorization = self._store.get(authorization_id)
                if authorization is not None:
                    authorization.state = "execution_unknown"
                return
            if outcome == "retry":
                authorization = self._store.get(authorization_id)
                if authorization is not None:
                    authorization.state = "awaiting_authorization"
                return
            self._store.pop(authorization_id, None)

    async def update(
        self,
        authorization_id: str,
        mutator: Callable[[PendingAuthorization], tuple[T, PendingAuthorization]],
    ) -> T:
        r"""以 compare-and-swap 方式原子更新一次授权。"""
        async with self._lock:
            authorization = self._store.get(authorization_id)
            if authorization is None:
                raise KeyError(authorization_id)
            value, updated = mutator(authorization)
            self._store[authorization_id] = updated
            return value

put async

Python
put(authorization: PendingAuthorization) -> None

保存一次挂起授权。

源代码位于： feishu/agent/session.py
Python
486
487
488
489
async def put(self, authorization: PendingAuthorization) -> None:
    r"""保存一次挂起授权。"""
    async with self._lock:
        self._store[authorization.authorization_id] = authorization

get async

Python
get(authorization_id: str) -> PendingAuthorization | None

按 authorization_id 读取挂起授权而不移除。

源代码位于： feishu/agent/session.py
Python
491
492
493
494
async def get(self, authorization_id: str) -> PendingAuthorization | None:
    r"""按 `authorization_id` 读取挂起授权而不移除。"""
    async with self._lock:
        return self._store.get(authorization_id)

pop async

Python
pop(authorization_id: str) -> PendingAuthorization | None

按 authorization_id 取出并移除一次挂起授权。

源代码位于： feishu/agent/session.py
Python
496
497
498
499
async def pop(self, authorization_id: str) -> PendingAuthorization | None:
    r"""按 `authorization_id` 取出并移除一次挂起授权。"""
    async with self._lock:
        return self._store.pop(authorization_id, None)

claim async

Python
claim(authorization_id: str) -> ClaimResult

原子地认领一次授权，返回 feishu.agent.session.ClaimResult。

源代码位于： feishu/agent/session.py
Python
501
502
503
504
505
506
507
508
509
510
async def claim(self, authorization_id: str) -> ClaimResult:
    r"""原子地认领一次授权，返回 [feishu.agent.session.ClaimResult][]。"""
    async with self._lock:
        authorization = self._store.get(authorization_id)
        if authorization is None:
            return ClaimResult.MISSING
        if authorization.state != "awaiting_authorization":
            return ClaimResult.ALREADY_CLAIMED
        authorization.state = "executing"
        return ClaimResult.CLAIMED

complete async

Python
complete(authorization_id: str, *, outcome: str) -> None

标记授权恢复的最终处置；retry 重开，unknown/frozen 冻结，其余终态移除。

源代码位于： feishu/agent/session.py
Python
512
513
514
515
516
517
518
519
520
521
522
523
524
525
async def complete(self, authorization_id: str, *, outcome: str) -> None:
    r"""标记授权恢复的最终处置；`retry` 重开，`unknown`/`frozen` 冻结，其余终态移除。"""
    async with self._lock:
        if outcome in ("unknown", "frozen"):
            authorization = self._store.get(authorization_id)
            if authorization is not None:
                authorization.state = "execution_unknown"
            return
        if outcome == "retry":
            authorization = self._store.get(authorization_id)
            if authorization is not None:
                authorization.state = "awaiting_authorization"
            return
        self._store.pop(authorization_id, None)

update async

Python
update(authorization_id: str, mutator: Callable[[PendingAuthorization], tuple[T, PendingAuthorization]]) -> T

以 compare-and-swap 方式原子更新一次授权。

源代码位于： feishu/agent/session.py
Python
527
528
529
530
531
532
533
534
535
536
537
538
539
async def update(
    self,
    authorization_id: str,
    mutator: Callable[[PendingAuthorization], tuple[T, PendingAuthorization]],
) -> T:
    r"""以 compare-and-swap 方式原子更新一次授权。"""
    async with self._lock:
        authorization = self._store.get(authorization_id)
        if authorization is None:
            raise KeyError(authorization_id)
        value, updated = mutator(authorization)
        self._store[authorization_id] = updated
        return value