toolregistry package

class toolregistry.ToolRegistry

Bases: object

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

Provides functionality to:

• Register and manage tools

• Merge multiple registries

• Execute tool calls

• Generate tool schemas

• Interface with MCP servers

_tools

Internal dictionary mapping tool names to Tool instances.

Type:

Dict[str, Tool]

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

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

Parameters:

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

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

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

Merge tools from another ToolRegistry into this one.

Handles name conflicts according to keep_existing parameter.

Parameters:

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

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

Raises:

TypeError – If other is not a ToolRegistry instance.

register_mcp_tools(server_url: str)

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.

Raises:

ImportError – If [mcp] extra is not installed.

async register_mcp_tools_async(server_url: str)

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.

Raises:

ImportError – If [mcp] extra is not installed.

register_openapi_tools(spec_url: str, base_url: str | None = None)

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.

Raises:

ImportError – If [openapi] extra is not installed.

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

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.

Raises:

ImportError – If [openapi] extra is not installed.

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

execute_tool_calls(tool_calls: List[Any]) → Dict[str, str]

Execute tool calls with optimized parallel/sequential execution.

Execution strategy:

• Sequential for 1-2 tool calls (avoids thread pool overhead)

• Parallel for 3+ tool calls (improves performance)

Parameters:

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

Returns:

Dictionary mapping tool call IDs to execution results.

Return type:

Dict[str, str]

Raises:

Exception – If any tool execution fails.

recover_tool_call_assistant_message(tool_calls: List[Any], 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[Any]) – 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]]

class toolregistry.Tool(*, name: str, description: str, parameters: Dict[str, Any], callable: Callable[[...], Any], is_async: bool = False, parameters_model: Any | None = None)

Bases: BaseModel

Base class representing an executable tool/function.

Provides core functionality for:

• Function wrapping and metadata management

• Parameter validation using Pydantic

• Synchronous/asynchronous execution

• JSON schema generation

name: str

The name of the tool.

Used as the primary identifier when calling the tool. Must be unique within a tool registry.

description: str

Detailed description of the tool’s functionality.

Should clearly explain what the tool does, its purpose, and any important usage considerations.

parameters: Dict[str, Any]

Parameter schema defining the tool’s expected inputs.

Follows JSON Schema format. Automatically generated from the wrapped function’s type hints when using from_function().

callable: Callable[[...], Any]

The underlying function/method that implements the tool’s logic.

This is excluded from serialization to prevent accidental exposure of sensitive implementation details.

is_async: bool

Flag indicating if the tool requires async execution.

Automatically detected from the wrapped function when using from_function(). Defaults to False for synchronous tools.

parameters_model: Any | None

Pydantic model used for parameter validation.

Automatically generated from the wrapped function’s type hints when using from_function(). Can be None for tools without parameter validation.

classmethod from_function(func: Callable[[...], Any], name: str | None = None, description: str | None = None) → Tool

Factory method to create Tool from callable.

Automatically:

• Extracts function metadata

• Generates parameter schema

• Handles async/sync detection

Parameters:

• func (Callable[..., Any]) – Function to convert to tool.

• name (Optional[str]) – Override tool name (defaults to function name).

• description (Optional[str]) – Override description (defaults to docstring).

Returns:

Configured Tool instance.

Return type:

Tool

Raises:

ValueError – For unnamed lambda functions.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

get_json_schema() → Dict[str, Any]

Generate JSON Schema representation of tool.

Schema includes:

• Name and description

• Parameter definitions

• Async flag

Returns:

JSON Schema compliant tool definition.

Return type:

Dict[str, Any]

describe() → Dict[str, Any]

Alias for get_json_schema.

Returns:

JSON schema representation of the tool

Return type:

Dict[str, Any]

run(parameters: Dict[str, Any]) → Any

Execute tool synchronously.

Parameters:

parameters (Dict[str, Any]) – Validated input parameters.

Returns:

Tool execution result.

Return type:

Any

Raises:

Exception – On execution failure.

async arun(parameters: Dict[str, Any]) → Any

Execute tool asynchronously.

Parameters:

parameters (Dict[str, Any]) – Validated input parameters.

Returns:

Tool execution result.

Return type:

Any

Raises:

• NotImplementedError – If async execution unsupported.

• Exception – On execution failure.

Submodules

• toolregistry.mcp_integration module
  • MCPToolWrapper
    • MCPToolWrapper.url
    • MCPToolWrapper.name
    • MCPToolWrapper.params
    • MCPToolWrapper.call_sync()
    • MCPToolWrapper.call_async()
  • MCPTool
    • MCPTool.name
    • MCPTool.description
    • MCPTool.parameters
    • MCPTool.callable
    • MCPTool.is_async
    • MCPTool.from_tool_json()
    • MCPTool.model_config
  • MCPIntegration
    • MCPIntegration.registry
    • MCPIntegration.register_mcp_tools_async()
    • MCPIntegration.register_mcp_tools()
• toolregistry.openapi_integration module
  • check_common_endpoints()
  • parse_openapi_spec_from_url()
  • get_openapi_spec()
  • OpenAPIToolWrapper
    • OpenAPIToolWrapper.base_url
    • OpenAPIToolWrapper.name
    • OpenAPIToolWrapper.method
    • OpenAPIToolWrapper.path
    • OpenAPIToolWrapper.params
    • OpenAPIToolWrapper.call_sync()
    • OpenAPIToolWrapper.call_async()
  • OpenAPITool
    • OpenAPITool.from_openapi_spec()
    • OpenAPITool.model_config
  • OpenAPIIntegration
    • OpenAPIIntegration.registry
    • OpenAPIIntegration.register_openapi_tools_async()
    • OpenAPIIntegration.register_openapi_tools()
• toolregistry.parameter_models module
  • InvalidSignature
    • InvalidSignature.message
  • ArgModelBase
    • ArgModelBase.model_config
    • ArgModelBase.model_dump_one_level()
    • ArgModelBase.model_config
• toolregistry.tool module
  • Tool
    • Tool.name
    • Tool.description
    • Tool.parameters
    • Tool.callable
    • Tool.is_async
    • Tool.parameters_model
    • Tool.from_function()
    • Tool.model_config
    • Tool.get_json_schema()
    • Tool.describe()
    • Tool.run()
    • Tool.arun()
• toolregistry.tool_registry module
  • ToolRegistry
    • ToolRegistry._tools
    • ToolRegistry.register()
    • ToolRegistry.merge()
    • ToolRegistry.register_mcp_tools()
    • ToolRegistry.register_mcp_tools_async()
    • ToolRegistry.register_openapi_tools()
    • ToolRegistry.register_openapi_tools_async()
    • ToolRegistry.get_available_tools()
    • ToolRegistry.get_tools_json()
    • ToolRegistry.get_tool()
    • ToolRegistry.get_callable()
    • ToolRegistry.execute_tool_calls()
    • ToolRegistry.recover_tool_call_assistant_message()