跳转至

tools

feishu.agent.tools

ToolValidationError

Bases: ValueError

工具参数校验失败时抛出。

当 feishu.agent.tools.ToolRegistry.dispatch 收到的参数不是对象、缺少必填字段，或在 additionalProperties 为 False 时出现多余字段，即抛出该异常。

示例：

Python Console Session
1
2
3
4
>>> raise ToolValidationError("missing required argument")
Traceback (most recent call last):
    ...
feishu.agent.tools.ToolValidationError: missing required argument
源代码位于： feishu/agent/tools.py
Python
33
34
35
36
37
38
39
40
41
42
43
44
45
class ToolValidationError(ValueError):
    r"""
    工具参数校验失败时抛出。

    当 [feishu.agent.tools.ToolRegistry.dispatch][] 收到的参数不是对象、缺少必填字段，或在
    `additionalProperties` 为 `False` 时出现多余字段，即抛出该异常。

    Examples:
        >>> raise ToolValidationError("missing required argument")
        Traceback (most recent call last):
            ...
        feishu.agent.tools.ToolValidationError: missing required argument
    """

Tool dataclass

一个已注册的工具：名称、描述、参数 Schema、处理函数及是否需要审批。

handler 既可为同步函数也可为协程函数；同步函数在分发时会被放到工作线程中执行，避免阻塞事件循环。 当 requires_approval 为 True 时，feishu.agent.loop.Agent 会先发送审批卡片并挂起本轮对话， 待用户批准后再执行。

示例：

Python Console Session
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
>>> async def weather(city):
...     return f"{city}：晴"
>>> tool = Tool(
...     name="weather",
...     description="查询天气",
...     input_schema={"type": "object", "properties": {"city": {"type": "string"}}},
...     handler=weather,
... )
>>> tool.requires_approval
False
源代码位于： feishu/agent/tools.py
Python
48
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
@dataclass
class Tool:
    r"""
    一个已注册的工具：名称、描述、参数 Schema、处理函数及是否需要审批。

    `handler` 既可为同步函数也可为协程函数；同步函数在分发时会被放到工作线程中执行，避免阻塞事件循环。
    当 `requires_approval` 为 `True` 时，[feishu.agent.loop.Agent][] 会先发送审批卡片并挂起本轮对话，
    待用户批准后再执行。

    Examples:
        >>> async def weather(city):
        ...     return f"{city}：晴"
        >>> tool = Tool(
        ...     name="weather",
        ...     description="查询天气",
        ...     input_schema={"type": "object", "properties": {"city": {"type": "string"}}},
        ...     handler=weather,
        ... )
        >>> tool.requires_approval
        False
    """

    name: str
    description: str
    input_schema: dict[str, Any]
    handler: Callable[..., Awaitable[Any] | Any]
    requires_approval: bool = False

ToolRegistry

工具注册表，负责工具的注册、声明导出与分发执行。

既支持装饰器形式注册，也支持直接传入处理函数；通过 feishu.agent.tools.ToolRegistry.specs 将 已注册工具导出为 feishu.agent.llm.ToolSpec 列表交给模型，再由 feishu.agent.tools.ToolRegistry.dispatch 校验参数并执行对应处理函数。

示例：

Python Console Session
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
>>> import asyncio
>>> reg = ToolRegistry()
>>> schema = {"type": "object"}
>>> async def weather(city):
...     return f"{city}：晴"
>>> _ = reg.register("weather", weather, input_schema=schema, description="天气")
>>> reg.specs()
[ToolSpec(name='weather', description='天气', input_schema={'type': 'object'})]
>>> asyncio.run(reg.dispatch("weather", {"city": "上海"}))
'上海：晴'
源代码位于： feishu/agent/tools.py
Python
 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
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
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
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
class ToolRegistry:
    r"""
    工具注册表，负责工具的注册、声明导出与分发执行。

    既支持装饰器形式注册，也支持直接传入处理函数；通过 [feishu.agent.tools.ToolRegistry.specs][] 将
    已注册工具导出为 [feishu.agent.llm.ToolSpec][] 列表交给模型，再由
    [feishu.agent.tools.ToolRegistry.dispatch][] 校验参数并执行对应处理函数。

    Examples:
        >>> import asyncio
        >>> reg = ToolRegistry()
        >>> schema = {"type": "object"}
        >>> async def weather(city):
        ...     return f"{city}：晴"
        >>> _ = reg.register("weather", weather, input_schema=schema, description="天气")
        >>> reg.specs()
        [ToolSpec(name='weather', description='天气', input_schema={'type': 'object'})]
        >>> asyncio.run(reg.dispatch("weather", {"city": "上海"}))
        '上海：晴'
    """

    def __init__(self) -> None:
        self._tools: dict[str, Tool] = {}

    def register(
        self,
        name: str | None = None,
        handler: Callable[..., Any] | None = None,
        *,
        input_schema: dict[str, Any],
        description: str,
        requires_approval: bool = False,
    ) -> Callable[..., Any] | None:
        r"""
        注册一个工具，支持装饰器与直接调用两种形式。

        直接传入 `handler` 时立即注册并原样返回该处理函数；省略 `handler` 时返回一个装饰器，可直接装饰处理
        函数。未显式指定 `name` 时取处理函数的 `__name__` 作为工具名。

        Args:
            name: 工具名称。省略时取处理函数的 `__name__`。
            handler: 工具处理函数，可为同步函数或协程函数。省略时本方法返回装饰器。
            input_schema: 描述工具参数的 JSON Schema。
            description: 工具描述，供模型理解其用途。
            requires_approval: 是否在执行前要求用户审批。默认为 `False`。

        Returns:
            直接调用形式下原样返回 `handler`；装饰器形式下返回用于装饰处理函数的装饰器。

        Raises:
            ValueError: 既未提供 `name` 又无法从处理函数推断出名称时抛出。

        Examples:
            >>> reg = ToolRegistry()
            >>> schema = {"type": "object", "properties": {}}
            >>> @reg.register("ping", input_schema=schema, description="心跳")
            ... async def ping():
            ...     return "pong"
            >>> reg.get("ping").name
            'ping'
        """

        def _add(fn: Callable[..., Any]) -> Callable[..., Any]:
            tool_name = name or getattr(fn, "__name__", None)
            if not tool_name:
                raise ValueError("tool name is required")
            self._tools[tool_name] = Tool(
                name=tool_name,
                description=description,
                input_schema=input_schema,
                handler=fn,
                requires_approval=requires_approval,
            )
            return fn

        if handler is not None:
            return _add(handler)
        return _add  # decorator form

    def specs(self) -> list[ToolSpec]:
        r"""
        将所有已注册工具导出为 [feishu.agent.llm.ToolSpec][] 列表。

        Returns:
            工具声明列表，可直接作为 `tools` 参数传给 [feishu.agent.llm.LlmBackend.stream][]。

        Examples:
            >>> reg = ToolRegistry()
            >>> async def ping():
            ...     return "pong"
            >>> _ = reg.register("ping", ping, input_schema={"type": "object"}, description="心跳")
            >>> reg.specs()
            [ToolSpec(name='ping', description='心跳', input_schema={'type': 'object'})]
        """
        return [
            ToolSpec(name=t.name, description=t.description, input_schema=t.input_schema) for t in self._tools.values()
        ]

    def get(self, name: str) -> Tool:
        r"""
        按名称获取已注册的工具。

        Args:
            name: 工具名称。

        Returns:
            对应的 [feishu.agent.tools.Tool][]。

        Raises:
            KeyError: 工具未注册时抛出。
        """
        return self._tools[name]

    async def dispatch(self, name: str, arguments: dict[str, Any]) -> Any:
        r"""
        校验参数并执行指定工具，返回其结果。

        先依据工具的 `input_schema` 校验 `arguments`，再调用对应处理函数。协程处理函数会被 `await`；
        同步处理函数则放到工作线程中执行，避免阻塞事件循环。

        Args:
            name: 工具名称。
            arguments: 已解析为字典的工具参数。

        Returns:
            工具处理函数的返回值。

        Raises:
            KeyError: 工具未注册时抛出。
            ToolValidationError: 参数未通过 `input_schema` 校验时抛出。

        Examples:
            >>> import asyncio
            >>> reg = ToolRegistry()
            >>> schema = {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}
            >>> async def weather(city):
            ...     return f"{city}：晴"
            >>> _ = reg.register("weather", weather, input_schema=schema, description="查询天气")
            >>> asyncio.run(reg.dispatch("weather", {"city": "北京"}))
            '北京：晴'
        """
        tool = self._tools[name]  # raises KeyError if unknown
        _validate(name, tool.input_schema, arguments)
        if inspect.iscoroutinefunction(tool.handler) or inspect.iscoroutinefunction(type(tool.handler).__call__):
            return await tool.handler(**arguments)
        result = await asyncio.to_thread(tool.handler, **arguments)
        if inspect.isawaitable(result):
            return await result
        return result

register

Python
register(name: str | None = None, handler: Callable[..., Any] | None = None, *, input_schema: dict[str, Any], description: str, requires_approval: bool = False) -> Callable[..., Any] | None

注册一个工具，支持装饰器与直接调用两种形式。

直接传入 handler 时立即注册并原样返回该处理函数；省略 handler 时返回一个装饰器，可直接装饰处理 函数。未显式指定 name 时取处理函数的 __name__ 作为工具名。

参数：

名称 类型 描述 默认
name
str | None

工具名称。省略时取处理函数的 __name__。

 None
handler
Callable[..., Any] | None

工具处理函数，可为同步函数或协程函数。省略时本方法返回装饰器。

 None
input_schema
dict[str, Any]

描述工具参数的 JSON Schema。

 必需
description
str

工具描述，供模型理解其用途。

 必需
requires_approval
bool

是否在执行前要求用户审批。默认为 False。

 False

返回：

类型 描述
Callable[..., Any] | None

直接调用形式下原样返回 handler；装饰器形式下返回用于装饰处理函数的装饰器。

引发：

类型 描述
ValueError

既未提供 name 又无法从处理函数推断出名称时抛出。

示例：

Python Console Session
1
2
3
4
5
6
7
>>> reg = ToolRegistry()
>>> schema = {"type": "object", "properties": {}}
>>> @reg.register("ping", input_schema=schema, description="心跳")
... async def ping():
...     return "pong"
>>> reg.get("ping").name
'ping'
源代码位于： feishu/agent/tools.py
Python
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def register(
    self,
    name: str | None = None,
    handler: Callable[..., Any] | None = None,
    *,
    input_schema: dict[str, Any],
    description: str,
    requires_approval: bool = False,
) -> Callable[..., Any] | None:
    r"""
    注册一个工具，支持装饰器与直接调用两种形式。

    直接传入 `handler` 时立即注册并原样返回该处理函数；省略 `handler` 时返回一个装饰器，可直接装饰处理
    函数。未显式指定 `name` 时取处理函数的 `__name__` 作为工具名。

    Args:
        name: 工具名称。省略时取处理函数的 `__name__`。
        handler: 工具处理函数，可为同步函数或协程函数。省略时本方法返回装饰器。
        input_schema: 描述工具参数的 JSON Schema。
        description: 工具描述，供模型理解其用途。
        requires_approval: 是否在执行前要求用户审批。默认为 `False`。

    Returns:
        直接调用形式下原样返回 `handler`；装饰器形式下返回用于装饰处理函数的装饰器。

    Raises:
        ValueError: 既未提供 `name` 又无法从处理函数推断出名称时抛出。

    Examples:
        >>> reg = ToolRegistry()
        >>> schema = {"type": "object", "properties": {}}
        >>> @reg.register("ping", input_schema=schema, description="心跳")
        ... async def ping():
        ...     return "pong"
        >>> reg.get("ping").name
        'ping'
    """

    def _add(fn: Callable[..., Any]) -> Callable[..., Any]:
        tool_name = name or getattr(fn, "__name__", None)
        if not tool_name:
            raise ValueError("tool name is required")
        self._tools[tool_name] = Tool(
            name=tool_name,
            description=description,
            input_schema=input_schema,
            handler=fn,
            requires_approval=requires_approval,
        )
        return fn

    if handler is not None:
        return _add(handler)
    return _add  # decorator form

specs

Python
specs() -> list[ToolSpec]

将所有已注册工具导出为 feishu.agent.llm.ToolSpec 列表。

返回：

类型 描述
list[ToolSpec]

工具声明列表，可直接作为 tools 参数传给 feishu.agent.llm.LlmBackend.stream。

示例：

Python Console Session
1
2
3
4
5
6
>>> reg = ToolRegistry()
>>> async def ping():
...     return "pong"
>>> _ = reg.register("ping", ping, input_schema={"type": "object"}, description="心跳")
>>> reg.specs()
[ToolSpec(name='ping', description='心跳', input_schema={'type': 'object'})]
源代码位于： feishu/agent/tools.py
Python
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def specs(self) -> list[ToolSpec]:
    r"""
    将所有已注册工具导出为 [feishu.agent.llm.ToolSpec][] 列表。

    Returns:
        工具声明列表，可直接作为 `tools` 参数传给 [feishu.agent.llm.LlmBackend.stream][]。

    Examples:
        >>> reg = ToolRegistry()
        >>> async def ping():
        ...     return "pong"
        >>> _ = reg.register("ping", ping, input_schema={"type": "object"}, description="心跳")
        >>> reg.specs()
        [ToolSpec(name='ping', description='心跳', input_schema={'type': 'object'})]
    """
    return [
        ToolSpec(name=t.name, description=t.description, input_schema=t.input_schema) for t in self._tools.values()
    ]

get

Python
get(name: str) -> Tool

按名称获取已注册的工具。

参数：

名称 类型 描述 默认
name
str

工具名称。

 必需

返回：

类型 描述
Tool

对应的 feishu.agent.tools.Tool。

引发：

类型 描述
KeyError

工具未注册时抛出。
源代码位于： feishu/agent/tools.py
Python
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def get(self, name: str) -> Tool:
    r"""
    按名称获取已注册的工具。

    Args:
        name: 工具名称。

    Returns:
        对应的 [feishu.agent.tools.Tool][]。

    Raises:
        KeyError: 工具未注册时抛出。
    """
    return self._tools[name]

dispatch async

Python
dispatch(name: str, arguments: dict[str, Any]) -> Any

校验参数并执行指定工具，返回其结果。

先依据工具的 input_schema 校验 arguments，再调用对应处理函数。协程处理函数会被 await； 同步处理函数则放到工作线程中执行，避免阻塞事件循环。

参数：

名称 类型 描述 默认
name
str

工具名称。

 必需
arguments
dict[str, Any]

已解析为字典的工具参数。

 必需

返回：

类型 描述
Any

工具处理函数的返回值。

引发：

类型 描述
KeyError

工具未注册时抛出。
ToolValidationError

参数未通过 input_schema 校验时抛出。

示例：

Python Console Session
1
2
3
4
5
6
7
8
>>> import asyncio
>>> reg = ToolRegistry()
>>> schema = {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}
>>> async def weather(city):
...     return f"{city}：晴"
>>> _ = reg.register("weather", weather, input_schema=schema, description="查询天气")
>>> asyncio.run(reg.dispatch("weather", {"city": "北京"}))
'北京：晴'
源代码位于： feishu/agent/tools.py
Python
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
async def dispatch(self, name: str, arguments: dict[str, Any]) -> Any:
    r"""
    校验参数并执行指定工具，返回其结果。

    先依据工具的 `input_schema` 校验 `arguments`，再调用对应处理函数。协程处理函数会被 `await`；
    同步处理函数则放到工作线程中执行，避免阻塞事件循环。

    Args:
        name: 工具名称。
        arguments: 已解析为字典的工具参数。

    Returns:
        工具处理函数的返回值。

    Raises:
        KeyError: 工具未注册时抛出。
        ToolValidationError: 参数未通过 `input_schema` 校验时抛出。

    Examples:
        >>> import asyncio
        >>> reg = ToolRegistry()
        >>> schema = {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}
        >>> async def weather(city):
        ...     return f"{city}：晴"
        >>> _ = reg.register("weather", weather, input_schema=schema, description="查询天气")
        >>> asyncio.run(reg.dispatch("weather", {"city": "北京"}))
        '北京：晴'
    """
    tool = self._tools[name]  # raises KeyError if unknown
    _validate(name, tool.input_schema, arguments)
    if inspect.iscoroutinefunction(tool.handler) or inspect.iscoroutinefunction(type(tool.handler).__call__):
        return await tool.handler(**arguments)
    result = await asyncio.to_thread(tool.handler, **arguments)
    if inspect.isawaitable(result):
        return await result
    return result