portia.tool_registry
A ToolRegistry represents a source of tools.
This module defines various implementations of ToolRegistry
, which is responsible for managing
and interacting with tools. It provides interfaces for registering, retrieving, and listing tools.
The ToolRegistry
can also support aggregation of multiple registries and searching for tools
based on queries.
Classes:
ToolRegistry: The base interface for managing tools.
AggregatedToolRegistry: A registry that aggregates multiple tool registries.
InMemoryToolRegistry: A simple in-memory implementation of ToolRegistry
.
PortiaToolRegistry: A tool registry that interacts with the Portia API to manage tools.
MCPToolRegistry: A tool registry that interacts with a locally running MCP server.
ToolRegistry Objects
class ToolRegistry(ABC)
ToolRegistry is the base interface for managing tools.
This class defines the essential methods for interacting with tool registries, including registering, retrieving, and listing tools. Specific tool registries should implement these methods.
Methods:
register_tool(tool
- Tool) -> None: Registers a new tool in the registry.get_tool(tool_id
- str) -> Tool: Retrieves a tool by its ID. get_tools() -> list[Tool]: Retrieves all tools in the registry.match_tools(query
- str | None = None, tool_ids: list[str] | None = None) -> list[Tool]: Optionally, retrieve tools that match a given query and tool_ids. Useful to implement tool filtering.
register_tool
@abstractmethod
def register_tool(tool: Tool) -> None
Register a new tool.
Arguments:
tool
Tool - The tool to be registered.
get_tool
@abstractmethod
def get_tool(tool_id: str) -> Tool
Retrieve a tool's information.
Arguments:
tool_id
str - The ID of the tool to retrieve.
Returns:
Tool
- The requested tool.
Raises:
ToolNotFoundError
- If the tool with the given ID does not exist.
get_tools
@abstractmethod
def get_tools() -> list[Tool]
Get all tools registered with the registry.
Returns:
list[Tool]
- A list of all tools in the registry.
match_tools
def match_tools(query: str | None = None,
tool_ids: list[str] | None = None) -> list[Tool]
Provide a set of tools that match a given query and tool_ids.
Arguments:
query
str | None - The query to match tools against.tool_ids
list[str] | None - The list of tool ids to match.
Returns:
-
list[Tool]
- A list of tools matching the query.This method is useful to implement tool filtering whereby only a selection of tools are passed to the PlanningAgent based on the query. This method is optional to implement and will default to providing all tools.
__add__
def __add__(other: ToolRegistry | list[Tool]) -> ToolRegistry
Return an aggregated tool registry combining two registries or a registry and tool list.
Tool IDs must be unique across the two registries otherwise an error will be thrown.
Arguments:
other
ToolRegistry - Another tool registry to be combined.
Returns:
AggregatedToolRegistry
- A new tool registry containing tools from both registries.
__radd__
def __radd__(other: ToolRegistry | list[Tool]) -> ToolRegistry
Return an aggregated tool registry combining two registries or a registry and tool list.
Tool IDs must be unique across the two registries otherwise an error will be thrown.
Arguments:
other
ToolRegistry - Another tool registry to be combined.
Returns:
AggregatedToolRegistry
- A new tool registry containing tools from both registries.
AggregatedToolRegistry Objects
class AggregatedToolRegistry(ToolRegistry)
An interface over a set of tool registries.
This class aggregates multiple tool registries, allowing the user to retrieve tools from any of the registries in the collection.
__init__
def __init__(registries: list[ToolRegistry]) -> None
Initialize the aggregated tool registry with a list of registries.
Arguments:
registries
list[ToolRegistry] - A list of tool registries to aggregate.
register_tool
def register_tool(tool: Tool) -> None
Throw not implemented error as registration should happen in individual registries.
get_tool
def get_tool(tool_id: str) -> Tool
Search across all registries for a given tool, returning the first match.
Arguments:
tool_id
str - The ID of the tool to retrieve.
Returns:
Tool
- The requested tool.
Raises:
ToolNotFoundError
- If the tool with the given ID does not exist in any registry.
get_tools
def get_tools() -> list[Tool]
Get all tools from all registries.
Returns:
list[Tool]
- A list of all tools across all registries.
match_tools
def match_tools(query: str | None = None,
tool_ids: list[str] | None = None) -> list[Tool]
Get all tools from all registries that match the query and tool_ids.
Arguments:
query
str | None - The query to match tools against.tool_ids
list[str] | None - The list of tool ids to match.
Returns:
list[Tool]
- A list of tools matching the query from all registries.
InMemoryToolRegistry Objects
class InMemoryToolRegistry(ToolRegistry)
Provides a simple in-memory tool registry.
This class stores tools in memory, allowing for quick access without persistence.
__init__
def __init__() -> None
Initialize the registry with an empty list of tools.
from_local_tools
@classmethod
def from_local_tools(cls, tools: Sequence[Tool]) -> InMemoryToolRegistry
Easily create a local tool registry from a sequence of tools.
Arguments:
tools
Sequence[Tool] - A sequence of tools to initialize the registry.
Returns:
InMemoryToolRegistry
- A new in-memory tool registry.
register_tool
def register_tool(tool: Tool) -> None
Register tool in the in-memory registry.
Arguments:
tool
Tool - The tool to register.
Raises:
DuplicateToolError
- If the tool ID already exists in the registry.
get_tool
def get_tool(tool_id: str) -> Tool
Get the tool from the in-memory registry.
Arguments:
tool_id
str - The ID of the tool to retrieve.
Returns:
Tool
- The requested tool.
Raises:
ToolNotFoundError
- If the tool with the given ID does not exist.
get_tools
def get_tools() -> list[Tool]
Get all tools in the in-memory registry.
Returns:
list[Tool]
- A list of all tools in the registry.
PortiaToolRegistry Objects
class PortiaToolRegistry(ToolRegistry)
Provides access to Portia tools.
This class interacts with the Portia API to retrieve and manage tools.
__init__
def __init__(config: Config,
tools: dict[str, Tool] | None = None,
client: httpx.Client | None = None) -> None
Initialize the PortiaToolRegistry with the given configuration.
Arguments:
config
Config - The configuration containing the API key and endpoint.tools
list[Tool] | None - A list of tools to create the registry with. If not provided, all tools will be loaded from the Portia API.client
httpx.Client | None - An optional httpx client to use. If not provided, a new client will be created.
with_default_tool_filter
@classmethod
def with_default_tool_filter(cls, config: Config) -> ToolRegistry
Create a PortiaToolRegistry with a default tool filter.
default_tool_filter
@classmethod
def default_tool_filter(cls, tool: Tool) -> bool
Filter to get the default set of tools offered by Portia cloud.
register_tool
def register_tool(tool: Tool) -> None
Throw not implemented error as registration can't be done in this registry.
get_tool
def get_tool(tool_id: str) -> Tool
Get the tool from the tool set.
Arguments:
tool_id
str - The ID of the tool to retrieve.
Returns:
Tool
- The requested tool.
Raises:
ToolNotFoundError
- If the tool with the given ID does not exist.
get_tools
def get_tools() -> list[Tool]
Get all tools in the registry.
Returns:
list[Tool]
- A list of all tools in the registry.
filter_tools
def filter_tools(filter_func: Callable[[Tool], bool]) -> ToolRegistry
Return a new registry with the tools filtered by the filter function.
McpToolRegistry Objects
class McpToolRegistry(ToolRegistry)
Provides access to tools within a Model Context Protocol (MCP) server.
See https://modelcontextprotocol.io/introduction for more information on MCP.
__init__
def __init__(mcp_client_config: McpClientConfig) -> None
Initialize the MCPToolRegistry with the given configuration.
from_sse_connection
@classmethod
def from_sse_connection(cls,
server_name: str,
url: str,
headers: dict[str, Any] | None = None,
timeout: float = 5,
sse_read_timeout: float = 60 * 5) -> McpToolRegistry
Create a new MCPToolRegistry using an SSE connection.
from_stdio_connection
@classmethod
def from_stdio_connection(
cls,
server_name: str,
command: str,
args: list[str] | None = None,
env: dict[str, str] | None = None,
encoding: str = "utf-8",
encoding_error_handler: Literal["strict", "ignore", "replace"] = "strict"
) -> McpToolRegistry
Create a new MCPToolRegistry using a stdio connection.
register_tool
def register_tool(tool: Tool) -> None
Register a new tool.
Arguments:
tool
Tool - The tool to be registered.
get_tool
def get_tool(tool_id: str) -> Tool
Retrieve a tool's information.
Arguments:
tool_id
str - The ID of the tool to retrieve.
Returns:
Tool
- The requested tool.
Raises:
ToolNotFoundError
- If the tool with the given ID does not exist.
get_tools
def get_tools() -> list[Tool]
Get all tools registered with the registry.
Returns:
list[Tool]
- A list of all tools in the registry.
DefaultToolRegistry Objects
class DefaultToolRegistry(AggregatedToolRegistry)
A registry providing a default set of tools.
This includes the following tools:
- All open source tools that don't require API keys
- Search tool if you have a Tavily API key
- Weather tool if you have an OpenWeatherMap API key
- Portia cloud tools if you have a Portia cloud API key
__init__
def __init__(config: Config) -> None
Initialize the default tool registry with the given configuration.
generate_pydantic_model_from_json_schema
def generate_pydantic_model_from_json_schema(
model_name: str, json_schema: dict[str, Any]) -> type[BaseModel]
Generate a Pydantic model based on a JSON schema.
Arguments:
model_name
str - The name of the Pydantic model.json_schema
dict[str, Any] - The schema to generate the model from.
Returns:
type[BaseModel]
- The generated Pydantic model class.