tools

feishu.agent.tools

ToolValidationError

Bases: ValueError

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

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

示例：

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

源代码位于： feishu/agent/tools.py

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.AgentEngine 会先发送审批卡片并挂起本轮对话， 待用户批准后再执行。

示例：

>>> 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

@dataclass
class Tool:
    r"""
    一个已注册的工具：名称、描述、参数 Schema、处理函数、是否需要审批，以及可预检的用户授权范围。

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

    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
    auth_scopes: tuple[str, ...] = ()

ToolRegistry

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

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

示例：

>>> 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": "上海"})).content
'上海：晴'

源代码位于： feishu/agent/tools.py

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": "上海"})).content
        '上海：晴'
    """

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

    @overload
    def register(
        self,
        name: str | None = None,
        handler: None = None,
        *,
        input_schema: dict[str, Any],
        description: str,
        requires_approval: bool = False,
        auth_scopes: Sequence[str] = (),
    ) -> Callable[[_ToolHandler], _ToolHandler]: ...

    @overload
    def register(
        self,
        name: str | None,
        handler: _ToolHandler,
        *,
        input_schema: dict[str, Any],
        description: str,
        requires_approval: bool = False,
        auth_scopes: Sequence[str] = (),
    ) -> _ToolHandler: ...

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

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

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

        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: _ToolHandler) -> _ToolHandler:
            tool_name = name or getattr(fn, "__name__", None)
            if not tool_name:
                raise ValueError("tool name is required")
            if tool_name in self._tools:
                raise ValueError(f"tool {tool_name!r} is already registered")
            self._tools[tool_name] = Tool(
                name=tool_name,
                description=description,
                input_schema=input_schema,
                handler=fn,
                requires_approval=requires_approval,
                auth_scopes=tuple(auth_scopes),
            )
            return fn

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

    def add(self, tool: Tool) -> Tool:
        r"""
        注册一个已构造的 [feishu.agent.tools.Tool][]，并原样返回。

        适用于注册由工厂产出的工具（如 [feishu.agent.toolkit][] 中的工厂），无需经 `register` 重新声明
        Schema 与描述。

        Args:
            tool: 待注册的工具。

        Returns:
            原样返回 `tool`，便于链式使用。

        Examples:
            >>> reg = ToolRegistry()
            >>> async def ping(): return "pong"
            >>> tool = Tool(name="ping", description="心跳", input_schema={"type": "object"}, handler=ping)
            >>> reg.add(tool).name
            'ping'
        """
        if tool.name in self._tools:
            raise ValueError(f"tool {tool.name!r} is already registered")
        self._tools[tool.name] = tool
        return tool

    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=tool.name, description=tool.description, input_schema=tool.input_schema)
            for tool 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]) -> ToolResult:
        r"""
        校验参数并执行指定工具，返回归一化的 [feishu.agent.result.ToolResult][]。

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

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

        Returns:
            归一化后的 [feishu.agent.result.ToolResult][]；处理函数返回的原始值会被包装为 `COMPLETED` 结果，
            使调用方（主循环、审批引擎）始终拿到统一的结果形状。

        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": "北京"})).content
            '北京：晴'
        """
        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__):
            result = await tool.handler(**arguments)
        else:
            result = await asyncio.to_thread(tool.handler, **arguments)
            if inspect.isawaitable(result):
                result = await result
        # Normalize every handler's return into a ToolResult so callers get one uniform shape; a raw value
        # becomes a COMPLETED result carrying it verbatim.
        return result if isinstance(result, ToolResult) else ToolResult(ToolOutcome.COMPLETED, content=result)

register

Python

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

Python

register(name: str | None, handler: _ToolHandler, *, input_schema: dict[str, Any], description: str, requires_approval: bool = False, auth_scopes: Sequence[str] = ()) -> _ToolHandler

Python

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

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

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

参数：

名称	类型	描述	默认
name	str | None	工具名称。省略时取处理函数的 __name__。	None
handler	_ToolHandler | None	工具处理函数，可为同步函数或协程函数。省略时本方法返回装饰器。	None
input_schema	dict[str, Any]	描述工具参数的 JSON Schema。	必需
description	str	工具描述，供模型理解其用途。	必需
requires_approval	bool	是否在执行前要求用户审批。默认为 False。	False
auth_scopes	Sequence[str]	工具执行前可预检的用户授权范围。默认为空。	()

返回：

类型	描述
Callable[[_ToolHandler], _ToolHandler] | _ToolHandler	直接调用形式下原样返回 handler；装饰器形式下返回用于装饰处理函数的装饰器。

引发：

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

示例：

>>> 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

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

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

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

    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: _ToolHandler) -> _ToolHandler:
        tool_name = name or getattr(fn, "__name__", None)
        if not tool_name:
            raise ValueError("tool name is required")
        if tool_name in self._tools:
            raise ValueError(f"tool {tool_name!r} is already registered")
        self._tools[tool_name] = Tool(
            name=tool_name,
            description=description,
            input_schema=input_schema,
            handler=fn,
            requires_approval=requires_approval,
            auth_scopes=tuple(auth_scopes),
        )
        return fn

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

add

Python

add(tool: Tool) -> Tool

注册一个已构造的 feishu.agent.tools.Tool，并原样返回。

适用于注册由工厂产出的工具（如 feishu.agent.toolkit 中的工厂），无需经 register 重新声明 Schema 与描述。

参数：

名称	类型	描述	默认
tool	Tool	待注册的工具。	必需

返回：

类型	描述
Tool	原样返回 tool，便于链式使用。

示例：

>>> reg = ToolRegistry()
>>> async def ping(): return "pong"
>>> tool = Tool(name="ping", description="心跳", input_schema={"type": "object"}, handler=ping)
>>> reg.add(tool).name
'ping'

源代码位于： feishu/agent/tools.py

def add(self, tool: Tool) -> Tool:
    r"""
    注册一个已构造的 [feishu.agent.tools.Tool][]，并原样返回。

    适用于注册由工厂产出的工具（如 [feishu.agent.toolkit][] 中的工厂），无需经 `register` 重新声明
    Schema 与描述。

    Args:
        tool: 待注册的工具。

    Returns:
        原样返回 `tool`，便于链式使用。

    Examples:
        >>> reg = ToolRegistry()
        >>> async def ping(): return "pong"
        >>> tool = Tool(name="ping", description="心跳", input_schema={"type": "object"}, handler=ping)
        >>> reg.add(tool).name
        'ping'
    """
    if tool.name in self._tools:
        raise ValueError(f"tool {tool.name!r} is already registered")
    self._tools[tool.name] = tool
    return tool

specs

Python

specs() -> list[ToolSpec]

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

返回：

类型	描述
list[ToolSpec]	工具声明列表，可直接作为 tools 参数传给 feishu.agent.llm.LlmBackend.stream。

示例：

>>> 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

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=tool.name, description=tool.description, input_schema=tool.input_schema)
        for tool in self._tools.values()
    ]

get

Python

get(name: str) -> Tool

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

参数：

名称	类型	描述	默认
name	str	工具名称。	必需

返回：

类型	描述
Tool	对应的 feishu.agent.tools.Tool。

引发：

类型	描述
KeyError	工具未注册时抛出。

源代码位于： feishu/agent/tools.py

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]) -> ToolResult

校验参数并执行指定工具，返回归一化的 feishu.agent.result.ToolResult。

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

参数：

名称	类型	描述	默认
name	str	工具名称。	必需
arguments	dict[str, Any]	已解析为字典的工具参数。	必需

返回：

类型	描述
ToolResult	归一化后的 feishu.agent.result.ToolResult；处理函数返回的原始值会被包装为 COMPLETED 结果，
ToolResult	使调用方（主循环、审批引擎）始终拿到统一的结果形状。

引发：

类型	描述
KeyError	工具未注册时抛出。
ToolValidationError	参数未通过 input_schema 校验时抛出。

示例：

>>> 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": "北京"})).content
'北京：晴'

源代码位于： feishu/agent/tools.py

async def dispatch(self, name: str, arguments: dict[str, Any]) -> ToolResult:
    r"""
    校验参数并执行指定工具，返回归一化的 [feishu.agent.result.ToolResult][]。

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

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

    Returns:
        归一化后的 [feishu.agent.result.ToolResult][]；处理函数返回的原始值会被包装为 `COMPLETED` 结果，
        使调用方（主循环、审批引擎）始终拿到统一的结果形状。

    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": "北京"})).content
        '北京：晴'
    """
    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__):
        result = await tool.handler(**arguments)
    else:
        result = await asyncio.to_thread(tool.handler, **arguments)
        if inspect.isawaitable(result):
            result = await result
    # Normalize every handler's return into a ToolResult so callers get one uniform shape; a raw value
    # becomes a COMPLETED result carrying it verbatim.
    return result if isinstance(result, ToolResult) else ToolResult(ToolOutcome.COMPLETED, content=result)