ToolRegistry¶
管理工具注册、执行和元数据的中央注册表类。
概述¶
ToolRegistry 是 ToolRegistry 库中工具管理的核心协调器。它提供了一个统一的接口,用于注册、发现和执行来自各种来源的工具,包括原生 Python 函数、OpenAPI 规范、MCP 服务器、LangChain 工具等。
核心特性¶
- 统一工具管理:所有类型工具的中央注册表
- 异步/同步支持:完全兼容同步和异步执行
- 命名空间组织:支持在命名空间下组织工具
- 多源集成:与各种工具源无缝集成
- 元数据保留:维护工具描述、参数和执行元数据
- 灵活执行:多种执行模式和并发选项
- 变更回调:通过
on_change()/remove_on_change()订阅工具状态变更 - 注册后钩子:通过
add_post_register_hook()在每个工具注册完成后运行自定义逻辑,支持自动禁用 - 基于标签的批量禁用:通过
disable_by_tags()按ToolTag值批量禁用多个工具
架构¶
ToolRegistry 遵循注册表模式,具有以下核心职责:
核心职责¶
- 工具注册:接受并注册来自各种来源的工具
- 工具发现:提供发现可用工具的机制
- 工具执行:使用适当的参数验证和错误处理执行工具
- 元数据管理:维护并提供对工具元数据的访问
- 命名空间支持:在逻辑命名空间下组织工具
注册方法¶
- 原生注册:
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
|
None
|
think_augment
|
bool
|
Enable thought-augmented tool calling globally.
When |
False
|
tool_discovery
|
bool
|
Enable tool discovery on initialization.
When |
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__ ¶
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__ ¶
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__ ¶
Return the JSON representation of the registry for debugging purposes.
Returns:
| Name | Type | Description |
|---|---|---|
str |
JSON string representation of the registry. |
__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: |
required |
api_format
|
API_FORMATS
|
Target API format. Defaults to |
'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_async
async
¶
Close all persistent connections (async).
Closes MCP and OpenAPI integrations that hold persistent connections or HTTP clients.
disable_think_augment ¶
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 and unregister the discovery tool.
enable_think_augment ¶
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 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: |
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 |
dict[str, str | list]
|
multimodal results (e.g. images). |
get_deferred_summaries ¶
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]]
|
|
list[dict[str, str | None]]
|
|
list[dict[str, str | None]]
|
|
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'
|
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
|
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 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:
|
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 ¶
Deprecated: use list_tools(include_disabled=True) instead.
list_tools ¶
List registered tools.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
include_disabled
|
bool
|
If |
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 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 ¶
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 应用程序中管理来自不同来源工具的中央枢纽。
另请参阅¶
- 事件 -
ChangeEvent、ChangeEventType和ChangeCallback的详细文档