ToolRegistry¶
The central registry class that manages tool registration, execution, and metadata across the ToolRegistry ecosystem.
Overview¶
ToolRegistry serves as the core orchestrator for tool management in the ToolRegistry library. It provides a unified interface for registering, discovering, and executing tools from various sources including native Python functions, OpenAPI specifications, MCP servers, LangChain tools, and more.
Key Features¶
- Unified Tool Management: Central registry for all types of tools
- Async/Sync Support: Full compatibility with both synchronous and asynchronous execution
- Namespace Organization: Support for organizing tools under namespaces
- Multi-Source Integration: Seamless integration with various tool sources
- Metadata Preservation: Maintains tool descriptions, parameters, and execution metadata
- Flexible Execution: Multiple execution modes and concurrency options
- Change Callbacks: Subscribe to tool state changes via
on_change()/remove_on_change() - Post-Registration Hooks: Run custom logic after each tool is registered via
add_post_register_hook(), with optional auto-disable support - Tag-Based Bulk Disable: Disable multiple tools at once by their
ToolTagvalues viadisable_by_tags()
Architecture¶
The ToolRegistry follows a registry pattern with the following key responsibilities:
Core Responsibilities¶
- Tool Registration: Accept and register tools from various sources
- Tool Discovery: Provide mechanisms to discover available tools
- Tool Execution: Execute tools with proper parameter validation and error handling
- Metadata Management: Maintain and provide access to tool metadata
- Namespace Support: Organize tools under logical namespaces
Registration Methods¶
- Native Registration:
register()for direct function/instance registration - Class Integration:
register_from_class()for Python class method registration. By default, traverses the MRO (Method Resolution Order) to include inherited methods from parent classes. Passtraverse_mro=Falseto register only directly defined methods. - OpenAPI Integration: Integration with OpenAPI specifications
- MCP Integration: Support for Model Context Protocol servers
- LangChain Integration: Compatibility with LangChain tools
Execution Models¶
invoke(tool_name, kwargs): Single-tool execution with full pipeline (permissions, logging, invocation tracking). Used by PTC for IPC callbacks.execute_tool_calls(tool_calls): Batch execution with concurrency via Thread/Process backends. Used for LLM tool_use responses.registry.ptc.enable(): Registers aprogrammatic_tool_calltool for Programmatic Tool Calling. LLMs can write Python code that orchestrates multiple tool calls.
Invocation Tracking¶
All tool executions are logged with an invocation_id prefix:
tr_sig_— singleinvoke()callstr_bat_— batchexecute_tool_calls()callstr_ptc_— PTC code execution tool calls
Query with log.get_entries(invocation_id="tr_ptc_...").
API Reference¶
toolregistry.ToolRegistry ¶
ToolRegistry(name: str | None = None, *, default_max_result_size: int | None = None, think_augment: bool = False, tool_discovery: bool = False, name_sep: Literal['-', '.'] = '-')
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
|
name_sep
|
Literal['-', '.']
|
Separator character used when combining namespace and
method name into a tool name (e.g. |
'-'
|
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.
ptc
property
¶
PTC (Programmatic Tool Calling) controller.
Use registry.ptc.enable() to register a code_execution
tool that lets LLMs write Python code with registered tools
callable in the namespace.
Example::
registry.ptc.enable(timeout=30)
registry.ptc.disable()
registry.ptc.enabled # bool
registry.ptc.last_invocation_id # str | None
__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. |
apply_metadata_config ¶
Apply tool_metadata overrides from a loaded ToolConfig.
Accepts dict[str, ToolMetadataOverride] (as returned by
ToolConfig.tool_metadata) or plain dict[str, dict].
Silently skips tool names not present in the registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
overrides
|
dict[str, Any]
|
Mapping of exact tool name → metadata override.
Each value must have |
required |
build_tool_call_messages ¶
build_tool_call_messages(tool_calls: list[Any], results: list[Any], 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.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tool_calls
|
list[Any]
|
Tool call objects in any supported format (as received from the LLM). |
required |
results
|
list[Any]
|
Structured results from :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[Any], execution_mode: Literal['process', 'thread'] | None = None) -> ResultList
Execute tool calls and return structured results.
Disabled tools are rejected with an :class:ErrorResult instead
of being executed. If logging is enabled, execution details are
recorded.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tool_calls
|
list[Any]
|
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 |
|---|---|
ResultList
|
List of results in the same order as tool_calls. Each |
ResultList
|
element is a :class: |
ResultList
|
class: |
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, } ]
invoke ¶
Execute a single tool with full pipeline (permissions, logging).
This is the canonical single-tool execution entry point. It is
used by :class:PtcTool for IPC callbacks and can be
called directly for programmatic tool invocation.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tool_name
|
str
|
Name of the registered tool. |
required |
kwargs
|
dict[str, Any]
|
Keyword arguments to pass to the tool. |
required |
invocation_id
|
str | None
|
Optional ID to group related calls. If
|
None
|
Returns:
| Type | Description |
|---|---|
Any
|
The tool's return value. |
Raises:
| Type | Description |
|---|---|
KeyError
|
If the tool is not registered. |
PermissionError
|
If the tool is denied by permission policy. |
RuntimeError
|
If the tool is disabled. |
Exception
|
Any exception raised by the tool itself. |
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[Any], results: list[Any], 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.
Usage Examples¶
Basic Tool Registration¶
from toolregistry import ToolRegistry
registry = ToolRegistry()
# Register a simple function
def add_numbers(a: int, b: int) -> int:
return a + b
registry.register(add_numbers)
Class Integration¶
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
# Register all methods from the class
registry.register_from_class(Calculator)
Class Integration with MRO Traversal¶
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()
# Default behavior (traverse_mro=True): includes inherited methods from BaseCalculator
registry.register_from_class(AdvancedCalculator)
print(registry.get_available_tools())
# Output: ['advanced_calculator-add', 'advanced_calculator-multiply']
# With traverse_mro=False: only methods defined directly on AdvancedCalculator
registry2 = ToolRegistry()
registry2.register_from_class(AdvancedCalculator, traverse_mro=False)
print(registry2.get_available_tools())
# Output: ['advanced_calculator-multiply']
Namespace Organization¶
from toolregistry import ToolRegistry
registry = ToolRegistry()
# Register with custom namespace
registry.register(my_function, namespace="math_utils")
# Access tools with namespace
available_tools = registry.get_available_tools(namespace="math_utils")
Change Callbacks¶
from toolregistry import ToolRegistry, ChangeEvent, ChangeEventType
registry = ToolRegistry()
def my_callback(event: ChangeEvent) -> None:
"""Handle tool registry changes."""
print(f"[{event.event_type.value}] {event.tool_name}")
if event.reason:
print(f" Reason: {event.reason}")
# Register the callback
registry.on_change(my_callback)
# Changes will trigger the callback
def add(a: int, b: int) -> int:
return a + b
registry.register(add) # Triggers: [register] add
registry.disable("add", reason="Maintenance") # Triggers: [disable] add
registry.enable("add") # Triggers: [enable] add
# Remove callback when no longer needed
registry.remove_on_change(my_callback)
Observability 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)
# Disable a tool with a reason
registry.disable("subtract", reason="Under maintenance")
# Get status of all tools
status = registry.get_tools_status()
print(status)
# Output:
# [
# {"name": "add", "enabled": True, "reason": None, "namespace": None},
# {"name": "subtract", "enabled": False, "reason": "Under maintenance", "namespace": None}
# ]
# Filter to find disabled tools
disabled_tools = [s for s in status if not s["enabled"]]
print(disabled_tools)
# Output: [{"name": "subtract", "enabled": False, "reason": "Under maintenance", "namespace": None}]
Tag-Based Bulk Disable¶
from toolregistry import ToolRegistry, ToolMetadata, ToolTag
registry = ToolRegistry()
def read_file(path: str) -> str:
"""Read a file from disk."""
...
def delete_file(path: str) -> None:
"""Delete a file from disk."""
...
def send_email(to: str, body: str) -> None:
"""Send an email."""
...
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" (default): disable tools that have AT LEAST ONE of the given tags
disabled = registry.disable_by_tags(
{ToolTag.DESTRUCTIVE, ToolTag.NETWORK},
match="any",
reason="Restricted in read-only mode",
)
print(disabled) # ['delete_file', 'send_email']
# match="all": disable only tools that carry EVERY specified tag
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="No destructive filesystem ops allowed",
)
print(disabled2) # ['delete_file']
Post-Registration Hook¶
from toolregistry import ToolRegistry, PostRegisterHook, ToolMetadata, ToolTag
registry = ToolRegistry()
# Hook: auto-disable any privileged tool at registration time
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"Privileged tool '{tool_name}' is not allowed in this environment"
return None
registry.add_post_register_hook(deny_privileged)
def sudo_command(cmd: str) -> str:
"""Run a command with elevated privileges."""
...
registry.register(sudo_command, metadata=ToolMetadata(tags={ToolTag.PRIVILEGED}))
print(registry.is_enabled("sudo_command")) # False
print(registry.get_disable_reason("sudo_command"))
# "Privileged tool 'sudo_command' is not allowed in this environment"
# Multiple hooks are invoked in registration order
def log_all(tool_name: str, tool, registry) -> None:
print(f"[hook] registered: {tool_name}")
registry.add_post_register_hook(log_all)
Integration Points¶
The ToolRegistry provides integration points for:
- OpenAPI Services: Automatic REST API tool generation
- MCP Servers: Model Context Protocol tool discovery
- LangChain Tools: LangChain ecosystem integration
- Native Python: Direct class and function registration
This makes it a central hub for managing tools from diverse sources within LLM applications.
See Also¶
- Events - Detailed documentation on
ChangeEvent,ChangeEventType, andChangeCallback