toolregistry.tool_registry module

class toolregistry.tool_registry.ToolRegistry(name: str | None = None)

Bases: object

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.

name

The name of the tool registry.

Type:

str

Notes

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

__init__(name: str | None = None) → None

Initialize an empty ToolRegistry.

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

Parameters:

name (Optional[str]) – Name of the tool registry. Defaults to a random “reg_<4-char>” string. For instance, “reg_1a3c”.

name

Name of the tool registry.

Type:

str

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.

register(tool_or_func: Callable | Tool, description: str | None = None, name: str | None = None, namespace: str | None = None)

Register a tool, either as a function, Tool instance, or static method.

Parameters:

• tool_or_func (Union[Callable, Tool]) – The tool to register, either as a function, Tool instance, or static method.

• description (Optional[str]) – Description for function tools. If not provided, the function’s docstring will be used.

• name (Optional[str]) – Custom name for the tool. If not provided, defaults to function name for functions or tool.name for Tool instances.

• namespace (Optional[str]) – Namespace for the tool. For static methods, defaults to class name if not provided.

merge(other: ToolRegistry, keep_existing: bool = False, force_namespace: bool = False)

Merge tools from another ToolRegistry into this one.

This method directly updates the current registry with tools from another registry, avoiding the need to create a new ToolRegistry object.

Parameters:

• other (ToolRegistry) – The ToolRegistry to merge from.

• keep_existing (bool) – If True, preserves existing tools on name conflicts.

• force_namespace (bool) – If True, forces updating tool namespaces by prefixing them with the registry name; if False, retains existing namespaces.

Raises:

TypeError – If other is not a ToolRegistry instance.

reduce_namespace() → None

Remove the namespace from tools in the registry if there is only one sub-registry.

This method checks if there is only one sub-registry remaining in the registry. If so, it removes the namespace prefix from all tools and clears the sub-registries.

Side Effects:

• Updates the _tools dictionary to remove namespace prefixes.

• Clears the _sub_registries set if namespace flattening occurs.

Example

If the registry contains tools with names like “calculator.add” and “calculator.subtract”, and “calculator” is the only sub-registry, this method will rename the tools to “add” and “subtract”.

spinoff(prefix: str, retain_namespace: bool = False) → ToolRegistry

Spin off tools with the specified prefix into a new registry.

This method creates a new ToolRegistry, transferring tools that belong to the specified prefix to it, and removing them from the current registry.

Parameters:

• prefix (str) – Prefix to identify tools to spin off.

• retain_namespace (bool) – If True, retains the namespace of tools in the current registry. If False, removes the namespace from tools after spinning off.

Returns:

A new registry containing the spun-off tools.

Return type:

ToolRegistry

Raises:

ValueError – If no tools with the specified prefix are found.

Notes

When retain_namespace is False, the reduce_namespace method is called to remove the namespace from tools in the current registry.

register_from_mcp(server_url: str, with_namespace: bool | str = False)

Register all tools from an MCP server (synchronous entry point).

Requires the [mcp] extra to be installed.

Parameters:

• server_url (str) – URL of the MCP server.

• with_namespace (Union[bool, str]) – Whether to prefix tool names with a namespace. - If False, no namespace is used. - If True, the namespace is derived from the OpenAPI info.title. - If a string is provided, it is used as the namespace. Defaults to False.

Raises:

ImportError – If [mcp] extra is not installed.

async register_from_mcp_async(server_url: str, with_namespace: bool | str = False)

Async implementation to register all tools from an MCP server.

Requires the [mcp] extra to be installed.

Parameters:

• server_url (str) – URL of the MCP server.

• with_namespace (Union[bool, str]) – Whether to prefix tool names with a namespace. - If False, no namespace is used. - If True, the namespace is derived from the OpenAPI info.title. - If a string is provided, it is used as the namespace. Defaults to False.

Raises:

ImportError – If [mcp] extra is not installed.

register_from_openapi(spec_url: str, base_url: str | None = None, with_namespace: bool | str = False)

Register all tools from an OpenAPI specification (synchronous entry point).

Requires the [openapi] extra to be installed.

Parameters:

• spec_url (str) – URL or path to the OpenAPI specification.

• base_url (Optional[str]) – Optional base URL to use if the spec does not provide a server.

• with_namespace (Union[bool, str]) – Whether to prefix tool names with a namespace. - If False, no namespace is used. - If True, the namespace is derived from the OpenAPI info.title. - If a string is provided, it is used as the namespace. Defaults to False.

Raises:

ImportError – If [openapi] extra is not installed.

async register_from_openapi_async(spec_url: str, base_url: str | None = None, with_namespace: bool | str = False)

Async implementation to register all tools from an OpenAPI specification.

Requires the [openapi] extra to be installed.

Parameters:

• spec_url (str) – URL or path to the OpenAPI specification.

• base_url (Optional[str]) – Optional base URL to use if the spec does not provide a server.

• with_namespace (Union[bool, str]) – Whether to prefix tool names with a namespace. - If False, no namespace is used. - If True, the namespace is derived from the OpenAPI info.title. - If a string is provided, it is used as the namespace. Defaults to False.

Raises:

ImportError – If [openapi] extra is not installed.

register_from_class(cls: Type | object, with_namespace: bool | str = False)

Register all static methods from a class or instance as tools.

Parameters:

• cls (Union[Type, object]) – The class or instance containing static methods to register.

• with_namespace (Union[bool, str]) – Whether to prefix tool names with a namespace. - If False, no namespace is used. - If True, the namespace is derived from the OpenAPI info.title. - If a string is provided, it is used as the namespace. Defaults to False.

Example

>>> from toolregistry.hub import Calculator
>>> registry = ToolRegistry()
>>> registry.register_from_class(Calculator)

Note

This method is now a convenience wrapper around the register() method’s static method handling capability.

async register_from_class_async(cls: Type | object, with_namespace: bool | str = False)

Async implementation to register all static methods from a class or instance as tools.

Parameters:

• cls (Union[Type, object]) – The class or instance containing static methods to register.

• with_namespace (Union[bool, str]) – Whether to prefix tool names with a namespace. - If False, no namespace is used. - If True, the namespace is derived from the OpenAPI info.title. - If a string is provided, it is used as the namespace. Defaults to False.

Example

>>> from toolregistry.hub import Calculator
>>> registry = ToolRegistry()
>>> registry.register_from_class(Calculator)

get_available_tools() → List[str]

List all registered tools.

Returns:

A list of tool names.

Return type:

List[str]

get_tools_json(tool_name: str | None = None) → List[Dict[str, Any]]

Get the JSON representation of all registered tools, following JSON Schema.

Parameters:

tool_name (Optional[str]) – Optional name of specific tool to get schema for.

Returns:

A list of tools in JSON format, compliant with JSON Schema.

Return type:

List[Dict[str, Any]]

get_tool(tool_name: str) → Tool | None

Get a tool by its name.

Parameters:

tool_name (str) – Name of the tool to retrieve.

Returns:

The tool, or None if not found.

Return type:

Optional[Tool]

get_callable(tool_name: str) → Callable[[...], Any] | None

Get a callable function by its name.

Parameters:

tool_name (str) – Name of the function to retrieve.

Returns:

The function to call, or None if not found.

Return type:

Optional[Callable[…, Any]]

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

Set the execution mode for parallel tasks.

Parameters:

mode (Literal["thread", "process"]) – The desired execution mode.

Raises:

ValueError – If an invalid mode is provided.

execute_tool_calls(tool_calls: List[ChatCompletionMessageToolCall], execution_mode: Literal['process', 'thread'] | None = None) → Dict[str, str]

Execute tool calls with concurrency using dill for serialization.

recover_tool_call_assistant_message(tool_calls: List[ChatCompletionMessageToolCall], tool_responses: Dict[str, str]) → List[Dict[str, Any]]

Construct assistant messages from tool call results.

Creates a conversation history with:

• Assistant tool call requests

• Tool execution responses

Parameters:

• tool_calls (List[ChatCompletionMessageToolCall]) – List of tool call objects.

• tool_responses (Dict[str, str]) – Dictionary of tool call IDs to results.

Returns:

List of message dictionaries in conversation format.

Return type:

List[Dict[str, Any]]

register_mcp_tools(server_url: str, with_namespace: bool | str = False)

async register_mcp_tools_async(server_url: str, with_namespace: bool | str = False)

register_openapi_tools(spec_url: str, base_url: str | None = None, with_namespace: bool | str = False)

async register_openapi_tools_async(spec_url: str, base_url: str | None = None, with_namespace: bool | str = False)

register_static_tools(cls: Type, with_namespace: bool | str = False)

async register_static_tools_async(cls: Type, with_namespace: bool | str = False)