跳转至

ToolRegistry

管理工具注册、执行和元数据的中央注册表类。

概述

ToolRegistry 是 ToolRegistry 库中工具管理的核心协调器。它提供了一个统一的接口,用于注册、发现和执行来自各种来源的工具,包括原生 Python 函数、OpenAPI 规范、MCP 服务器、LangChain 工具等。

核心特性

  • 统一工具管理:所有类型工具的中央注册表
  • 异步/同步支持:完全兼容同步和异步执行
  • 命名空间组织:支持在命名空间下组织工具
  • 多源集成:与各种工具源无缝集成
  • 元数据保留:维护工具描述、参数和执行元数据
  • 灵活执行:多种执行模式和并发选项
  • 变更回调:通过 on_change() / remove_on_change() 订阅工具状态变更
  • 注册后钩子:通过 add_post_register_hook() 在每个工具注册完成后运行自定义逻辑,支持自动禁用
  • 基于标签的批量禁用:通过 disable_by_tags()ToolTag 值批量禁用多个工具

架构

ToolRegistry 遵循注册表模式,具有以下核心职责:

核心职责

  1. 工具注册:接受并注册来自各种来源的工具
  2. 工具发现:提供发现可用工具的机制
  3. 工具执行:使用适当的参数验证和错误处理执行工具
  4. 元数据管理:维护并提供对工具元数据的访问
  5. 命名空间支持:在逻辑命名空间下组织工具

注册方法

  • 原生注册register() 用于直接函数/实例注册
  • 类集成register_from_class() 用于 Python 类方法注册。默认情况下,遍历 MRO(方法解析顺序)以包含从父类继承的方法。传递 traverse_mro=False 仅注册直接定义的方法。
  • OpenAPI 集成:与 OpenAPI 规范集成
  • MCP 集成:支持模型上下文协议服务器
  • LangChain 集成:与 LangChain 工具兼容

执行模型

  • 同步执行:非异步工具的直接执行
  • 异步执行:异步工具的 async/await 支持
  • 并发执行:支持并行工具执行
  • 错误处理:全面的错误处理和日志记录

API 参考

toolregistry.ToolRegistry

ToolRegistry(name: str | None = None, *, default_max_result_size: int | None = None, think_augment: bool = False, tool_discovery: bool = False)

Bases: AdminMixin, ExecutionLoggingMixin, PermissionsMixin, RegistrationMixin, EnableDisableMixin, NamespaceMixin, ChangeCallbackMixin

Central registry for managing tools (functions) and their metadata.

This class provides functionality to register, manage, and execute tools, as well as to interface with MCP servers, OpenAPI endpoints, and generate tool schemas.

Attributes:

Name Type Description
name str

The name of the tool registry.

Notes

Private attributes are used internally to manage registered tools and sub-registries. These attributes are not intended for external use.

Initialize an empty ToolRegistry.

This method initializes an empty ToolRegistry with a name and internal structures for storing tools and sub-registries.

Parameters:

Name Type Description Default
name str | None

Name of the tool registry. Defaults to a random "reg_<4-char>" string. For instance, "reg_1a3c".

None
default_max_result_size int | None

Default maximum result size in characters for all tools. Individual tools can override this via ToolMetadata.max_result_size. None means no limit.

None
think_augment bool

Enable thought-augmented tool calling globally. When True, a toolcall_reason property is included in every tool's schema so LLMs can articulate their rationale when calling tools. Individual tools can override this via ToolMetadata.think_augment. Defaults to False.

False
tool_discovery bool

Enable tool discovery on initialization. When True, :meth:enable_tool_discovery is called automatically, registering a discover_tools tool that LLMs can use to discover other tools by exact name or natural language query. Defaults to False.

False
Notes

This class uses private attributes _tools and _sub_registries internally to manage registered tools and sub-registries. These are not intended for external use.

__contains__

__contains__(name: str) -> bool

Check if a tool with the given name is registered.

Parameters:

Name Type Description Default
name str

Name of the tool to check.

required

Returns:

Name Type Description
bool bool

True if tool is registered, False otherwise.

__getitem__

__getitem__(key: str) -> Callable[..., Any] | None

Enable key-value access to retrieve callables.

Parameters:

Name Type Description Default
key str

Name of the function.

required

Returns:

Type Description
Callable[..., Any] | None

Optional[Callable[..., Any]]: The function to call, or None if not found.

__repr__

__repr__()

Return the JSON representation of the registry for debugging purposes.

Returns:

Name Type Description
str

JSON string representation of the registry.

__str__

__str__()

Return the JSON representation of the registry as a string.

Returns:

Name Type Description
str

JSON string representation of the registry.

build_tool_call_messages

build_tool_call_messages(tool_calls: list[AnyToolCall], tool_responses: dict[str, str | list], api_format: API_FORMATS = 'openai-chat') -> list[dict[str, Any]]

Build conversation messages for a tool-calling round-trip.

Combines the assistant message (tool call requests) and the tool result messages into the format required by the next LLM turn.

This is a convenience method wrapping :func:build_assistant_message and :func:build_tool_response. It handles Gemini-specific ID alignment automatically (position-based remapping).

.. important::

Do **not** reorder ``tool_calls`` between
:meth:`execute_tool_calls` and this method.  Gemini format
relies on positional alignment between ``tool_calls`` and
``tool_responses`` because Gemini does not provide tool call
IDs upstream.

Parameters:

Name Type Description Default
tool_calls list[AnyToolCall]

Tool call objects in any supported format.

required
tool_responses dict[str, str | list]

Mapping of tool call IDs to results, as returned by :meth:execute_tool_calls. Values are str or list[ContentBlock] for multimodal results.

required
api_format API_FORMATS

Target API format. Defaults to "openai-chat".

'openai-chat'

Returns:

Type Description
list[dict[str, Any]]

Conversation messages ready to extend the message history.

list[dict[str, Any]]

When multimodal content is present, an additional user

list[dict[str, Any]]

message is appended containing the expanded content.

close

close() -> None

Close all persistent connections (sync).

close_async async

close_async() -> None

Close all persistent connections (async).

Closes MCP and OpenAPI integrations that hold persistent connections or HTTP clients.

disable_think_augment

disable_think_augment() -> None

Disable thought-augmented tool calling globally.

When disabled, the toolcall_reason property is stripped from tool schemas produced by :meth:get_schemas, unless a tool explicitly opts in via ToolMetadata.think_augment = True.

disable_tool_discovery

disable_tool_discovery() -> None

Disable tool discovery and unregister the discovery tool.

enable_think_augment

enable_think_augment() -> None

Enable thought-augmented tool calling globally.

When enabled, a toolcall_reason property is included in every tool's schema (via :meth:get_schemas) so that LLMs can articulate their rationale when calling tools. Individual tools can still override this via ToolMetadata.think_augment.

Reference: https://arxiv.org/abs/2601.18282

enable_tool_discovery

enable_tool_discovery(field_weights: dict[str, float] | None = None) -> ToolDiscoveryTool

Enable tool discovery and register a discovery tool.

Creates a :class:ToolDiscoveryTool, registers its :meth:~ToolDiscoveryTool.discover method as a callable tool named discover_tools, and subscribes to registry change events for automatic index rebuilds.

The discovery tool itself is never deferred (defer=False) so that LLMs always see it in the initial schema.

Parameters:

Name Type Description Default
field_weights dict[str, float] | None

Optional per-field BM25F boost weights.

None

Returns:

Name Type Description
The ToolDiscoveryTool

class:ToolDiscoveryTool instance.

execute_tool_calls

execute_tool_calls(tool_calls: list[AnyToolCall], execution_mode: Literal['process', 'thread'] | None = None) -> dict[str, str | list]

Execute tool calls with concurrency using cloudpickle for serialization.

Disabled tools are rejected with an error message instead of being executed. If logging is enabled, execution details are recorded.

Parameters:

Name Type Description Default
tool_calls list[AnyToolCall]

List of tool calls to be executed in any supported format.

required
execution_mode Literal['process', 'thread'] | None

Execution mode to use; defaults to the Executor's current mode.

None

Returns:

Type Description
dict[str, str | list]

Dictionary mapping tool call IDs to their results. Values

dict[str, str | list]

are str for normal results or list[ContentBlock] for

dict[str, str | list]

multimodal results (e.g. images).

get_available_tools

get_available_tools() -> list[str]

Deprecated: use :meth:list_tools instead.

get_deferred_summaries

get_deferred_summaries() -> list[dict[str, str | None]]

Get name and first-sentence description for deferred tools.

Useful for injecting into system prompts so the LLM knows which additional tools are available via discover_tools.

Only enabled tools with ToolMetadata.defer=True are included.

Returns:

Type Description
list[dict[str, str | None]]

List of dicts with keys:

list[dict[str, str | None]]
  • name (str): Tool name.
list[dict[str, str | None]]
  • description (str): First sentence of the tool description.
list[dict[str, str | None]]
  • namespace (str | None): Tool namespace, if any.

get_schemas

get_schemas(tool_name: str | None = None, *, api_format: API_FORMATS = 'openai-chat', tags: set[str | ToolTag] | None = None, exclude_tags: set[str | ToolTag] | None = None, sort: bool = True, include_deferred: bool = True) -> list[dict[str, Any]]

Get tool definitions as JSON Schema dicts for a target API format.

When no specific tool_name is given, only enabled tools are returned. Tools can be filtered by tags and sorted for deterministic ordering.

Parameters:

Name Type Description Default
tool_name str | None

Optional name of specific tool to get schema for. When set, tag filtering and sorting are skipped.

None
api_format API_FORMATS

Target API format. Defaults to "openai-chat".

'openai-chat'
tags set[str | ToolTag] | None

If set, only include tools matching ANY of these tags.

None
exclude_tags set[str | ToolTag] | None

Exclude tools matching ANY of these tags.

None
sort bool

If True (default), sort tools by name for deterministic ordering. Stable sorting improves prompt cache hit rates.

True
include_deferred bool

If False, exclude tools with metadata.defer == True. Defaults to True for backward compatibility. Set to False when tool search is enabled so that deferred tools are only discovered via search.

True

Returns:

Type Description
list[dict[str, Any]]

A list of tool definition dicts in the specified API format.

get_tools_json

get_tools_json(tool_name: str | None = None, *, api_format: API_FORMATS = 'openai-chat', tags: set[str | ToolTag] | None = None, exclude_tags: set[str | ToolTag] | None = None, sort: bool = True, include_deferred: bool = True) -> list[dict[str, Any]]

Deprecated: use :meth:get_schemas instead.

get_tools_status

get_tools_status() -> list[dict[str, Any]]

Get status information for all registered tools.

Returns a list of dictionaries containing status information for each tool, including enable/disable state, metadata summary, and tags.

Returns:

Type Description
list[dict[str, Any]]

list[dict[str, Any]]: List of tool status dictionaries, each containing:

  • name (str): Tool name (with namespace prefix if applicable)
  • enabled (bool): Whether the tool is currently enabled
  • reason (str | None): Reason for disabling, if disabled
  • namespace (str | None): Namespace the tool belongs to
  • tags (list[str]): Sorted union of predefined and custom tags
  • locality (str): "local", "remote", or "any"
  • is_async (bool): Whether the tool requires async execution
  • source (str): Origin of the tool (e.g. "native", "mcp", "openapi", "langchain")
  • source_detail (str): Extra detail about the tool's origin (e.g. transport URI, spec URL, class name)
  • think_augment (bool | None): Think-augmented calling setting
  • defer (bool): Whether the tool is deferred from initial prompt
Example

registry = ToolRegistry() registry.register(my_tool) registry.disable("my_tool", reason="Under maintenance") registry.get_tools_status() [ { "name": "my_tool", "enabled": False, "reason": "Under maintenance", "namespace": None, "tags": [], "locality": "any", "is_async": False, "think_augment": None, "defer": False, } ]

list_all_tools

list_all_tools() -> list[str]

Deprecated: use list_tools(include_disabled=True) instead.

list_tools

list_tools(include_disabled: bool = False) -> list[str]

List registered tools.

Parameters:

Name Type Description Default
include_disabled bool

If True, include disabled tools in the result. Defaults to False (only enabled tools).

False

Returns:

Type Description
list[str]

List[str]: A list of tool names.

recover_tool_call_assistant_message

recover_tool_call_assistant_message(tool_calls: list[AnyToolCall], tool_responses: dict[str, str | list], api_format: API_FORMATS = 'openai-chat') -> list[dict[str, Any]]

Deprecated: use :meth:build_tool_call_messages instead.

set_default_execution_mode

set_default_execution_mode(mode: Literal['thread', 'process']) -> None

Set the default execution mode for parallel tasks.

This sets the default mode used by :meth:execute_tool_calls when no per-call execution_mode override is provided.

Parameters:

Name Type Description Default
mode Literal['thread', 'process']

The desired execution mode.

required

Raises:

Type Description
ValueError

If an invalid mode is provided.

set_execution_mode

set_execution_mode(mode: Literal['thread', 'process']) -> None

Deprecated: use :meth:set_default_execution_mode instead.

使用示例

基本工具注册

from toolregistry import ToolRegistry

registry = ToolRegistry()

# 注册一个简单函数
def add_numbers(a: int, b: int) -> int:
    return a + b

registry.register(add_numbers)

类集成

from toolregistry import ToolRegistry

registry = ToolRegistry()

class Calculator:
    @staticmethod
    def multiply(a: int, b: int) -> int:
        return a * b

    def divide(self, a: int, b: int) -> float:
        return a / b

# 注册类中的所有方法
registry.register_from_class(Calculator)

带 MRO 遍历的类集成

from toolregistry import ToolRegistry

class BaseCalculator:
    @staticmethod
    def add(a: int, b: int) -> int:
        return a + b

class AdvancedCalculator(BaseCalculator):
    @staticmethod
    def multiply(a: int, b: int) -> int:
        return a * b

registry = ToolRegistry()

# 默认行为(traverse_mro=True):包含从 BaseCalculator 继承的方法
registry.register_from_class(AdvancedCalculator)
print(registry.get_available_tools())
# 输出:['advanced_calculator-add', 'advanced_calculator-multiply']

# 使用 traverse_mro=False:仅注册直接定义在 AdvancedCalculator 上的方法
registry2 = ToolRegistry()
registry2.register_from_class(AdvancedCalculator, traverse_mro=False)
print(registry2.get_available_tools())
# 输出:['advanced_calculator-multiply']

命名空间组织

from toolregistry import ToolRegistry

registry = ToolRegistry()

# 使用自定义命名空间注册
registry.register(my_function, namespace="math_utils")

# 使用命名空间访问工具
available_tools = registry.get_available_tools(namespace="math_utils")

变更回调

from toolregistry import ToolRegistry, ChangeEvent, ChangeEventType

registry = ToolRegistry()

def my_callback(event: ChangeEvent) -> None:
    """处理工具注册表变更。"""
    print(f"[{event.event_type.value}] {event.tool_name}")
    if event.reason:
        print(f"  原因:{event.reason}")

# 注册回调
registry.on_change(my_callback)

# 变更将触发回调
def add(a: int, b: int) -> int:
    return a + b

registry.register(add)  # 触发:[register] add
registry.disable("add", reason="维护中")  # 触发:[disable] add
registry.enable("add")  # 触发:[enable] add

# 不再需要时移除回调
registry.remove_on_change(my_callback)

可观测性 API

from toolregistry import ToolRegistry

registry = ToolRegistry()

def add(a: int, b: int) -> int:
    return a + b

def subtract(a: int, b: int) -> int:
    return a - b

registry.register(add)
registry.register(subtract)

# 禁用一个工具并提供原因
registry.disable("subtract", reason="维护中")

# 获取所有工具的状态
status = registry.get_tools_status()
print(status)
# 输出:
# [
#     {"name": "add", "enabled": True, "reason": None, "namespace": None},
#     {"name": "subtract", "enabled": False, "reason": "维护中", "namespace": None}
# ]

# 筛选出已禁用的工具
disabled_tools = [s for s in status if not s["enabled"]]
print(disabled_tools)
# 输出:[{"name": "subtract", "enabled": False, "reason": "维护中", "namespace": None}]

基于标签的批量禁用

from toolregistry import ToolRegistry, ToolMetadata, ToolTag

registry = ToolRegistry()

def read_file(path: str) -> str:
    """从磁盘读取文件。"""
    ...

def delete_file(path: str) -> None:
    """从磁盘删除文件。"""
    ...

def send_email(to: str, body: str) -> None:
    """发送电子邮件。"""
    ...

registry.register(read_file, metadata=ToolMetadata(tags={ToolTag.FILE_SYSTEM, ToolTag.READ_ONLY}))
registry.register(delete_file, metadata=ToolMetadata(tags={ToolTag.FILE_SYSTEM, ToolTag.DESTRUCTIVE}))
registry.register(send_email, metadata=ToolMetadata(tags={ToolTag.NETWORK}))

# match="any"(默认):工具拥有至少一个指定标签即被禁用
disabled = registry.disable_by_tags(
    {ToolTag.DESTRUCTIVE, ToolTag.NETWORK},
    match="any",
    reason="只读模式下限制访问",
)
print(disabled)  # ['delete_file', 'send_email']

# match="all":工具携带全部指定标签才被禁用
registry2 = ToolRegistry()
registry2.register(read_file, metadata=ToolMetadata(tags={ToolTag.FILE_SYSTEM, ToolTag.READ_ONLY}))
registry2.register(delete_file, metadata=ToolMetadata(tags={ToolTag.FILE_SYSTEM, ToolTag.DESTRUCTIVE}))

disabled2 = registry2.disable_by_tags(
    {ToolTag.FILE_SYSTEM, ToolTag.DESTRUCTIVE},
    match="all",
    reason="不允许破坏性文件系统操作",
)
print(disabled2)  # ['delete_file']

注册后钩子

from toolregistry import ToolRegistry, PostRegisterHook, ToolMetadata, ToolTag

registry = ToolRegistry()

# 钩子:在注册时自动禁用所有特权工具
def deny_privileged(tool_name: str, tool, registry) -> str | None:
    tags = tool.metadata.tags if tool.metadata else set()
    if ToolTag.PRIVILEGED in tags:
        return f"当前环境不允许特权工具 '{tool_name}'"
    return None

registry.add_post_register_hook(deny_privileged)

def sudo_command(cmd: str) -> str:
    """以提升的权限运行命令。"""
    ...

registry.register(sudo_command, metadata=ToolMetadata(tags={ToolTag.PRIVILEGED}))

print(registry.is_enabled("sudo_command"))     # False
print(registry.get_disable_reason("sudo_command"))
# "当前环境不允许特权工具 'sudo_command'"

# 可以注册多个钩子,按注册顺序依次调用
def log_all(tool_name: str, tool, registry) -> None:
    print(f"[钩子] 已注册:{tool_name}")

registry.add_post_register_hook(log_all)

集成点

ToolRegistry 提供以下集成点:

  • OpenAPI 服务:自动 REST API 工具生成
  • MCP 服务器:模型上下文协议工具发现
  • LangChain 工具:LangChain 生态系统集成
  • 原生 Python:直接类和函数注册

这使其成为在 LLM 应用程序中管理来自不同来源工具的中央枢纽。

另请参阅

  • 事件 - ChangeEventChangeEventTypeChangeCallback 的详细文档