跳转至

session

feishu.agent.session

SessionStore

Bases: Protocol

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

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

示例：

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

    [feishu.agent.loop.Agent][] 通过该协议读写各会话的对话历史；内置实现为
    [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"""以给定的消息列表整体替换指定会话的历史。"""
        ...

get async

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

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

源代码位于： feishu/agent/session.py
Python
45
46
47
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
49
50
51
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
53
54
55
async def set(self, session_id: str, messages: list[Message]) -> None:
    r"""以给定的消息列表整体替换指定会话的历史。"""
    ...

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
 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
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._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)

    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)

get async

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

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

参数：

名称 类型 描述 默认
session_id
str

会话标识。

 必需

返回：

类型 描述
list[Message]

历史消息的副本；会话不存在时返回空列表。对返回列表的修改不会影响内部状态。
源代码位于： feishu/agent/session.py
Python
81
82
83
84
85
86
87
88
89
90
91
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
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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)

set async

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

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

参数：

名称 类型 描述 默认
session_id
str

会话标识。

 必需
messages
list[Message]

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

 必需
源代码位于： feishu/agent/session.py
Python
107
108
109
110
111
112
113
114
115
116
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)

PendingApproval dataclass

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

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

示例：

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
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
@dataclass
class PendingApproval:
    r"""
    一次挂起的工具审批，记录恢复对话所需的全部上下文。

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

    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]

PendingApprovalStore

Bases: Protocol

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

feishu.agent.loop.Agent 通过该协议保存与取回挂起的 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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@runtime_checkable
class PendingApprovalStore(Protocol):
    r"""
    挂起审批存储协议，是自定义审批持久化后端的扩展契约。

    [feishu.agent.loop.Agent][] 通过该协议保存与取回挂起的 [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`。"""
        ...

put async

Python
put(approval: PendingApproval) -> None

保存一次挂起的审批。

源代码位于： feishu/agent/session.py
Python
160
161
162
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
164
165
166
async def pop(self, approval_id: str) -> PendingApproval | None:
    r"""按 `approval_id` 取出并移除一次挂起的审批，不存在时返回 `None`。"""
    ...

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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
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)

put async

Python
put(approval: PendingApproval) -> None

保存一次挂起的审批。

参数：

名称 类型 描述 默认
approval
PendingApproval

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

 必需
源代码位于： feishu/agent/session.py
Python
199
200
201
202
203
204
205
206
207
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
209
210
211
212
213
214
215
216
217
218
219
220
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)