Tools

Define tools and register them for an agent to call.

Tool

Tool

Bases: BaseModel

Concrete tool combining definition, metadata, and executor.

to_openai_schema

to_openai_schema() -> dict[str, Any]

Convert to OpenAI function calling format.

Source code in redis_agent_kit/tools/models.py

def to_openai_schema(self) -> dict[str, Any]:
    """Convert to OpenAI function calling format."""
    return self.definition.to_openai_schema()

ToolDefinition

ToolDefinition

Bases: BaseModel

Definition of a tool that the LLM can call.

This describes the tool schema in OpenAI function calling format.

to_openai_schema

to_openai_schema() -> dict[str, Any]

Convert to OpenAI function calling format.

Source code in redis_agent_kit/tools/models.py

def to_openai_schema(self) -> dict[str, Any]:
    """Convert to OpenAI function calling format."""
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        },
    }

ToolManager

ToolManager

ToolManager()

Manages tool providers and routes tool calls.

Example

manager = ToolManager() manager.add_provider(WeatherProvider()) manager.add_provider(KnowledgeProvider(knowledge_store))

Get all tools in OpenAI format

tools = manager.get_openai_tools()

Execute a tool call

result = await manager.execute("get_weather", {"location": "NYC"})

Initialize tool manager.

Source code in redis_agent_kit/tools/manager.py

def __init__(self) -> None:
    """Initialize tool manager."""
    self._providers: dict[str, ToolProvider] = {}
    self._routing_table: dict[str, tuple[ToolProvider, Tool]] = {}
    self._tools: list[Tool] = []

providers property

providers: dict[str, ToolProvider]

Get all registered providers.

add_provider

add_provider(provider: ToolProvider) -> None

Add a tool provider.

Parameters:

Name	Type	Description	Default
provider	ToolProvider	ToolProvider instance to add.	required

Source code in redis_agent_kit/tools/manager.py

def add_provider(self, provider: ToolProvider) -> None:
    """Add a tool provider.

    Args:
        provider: ToolProvider instance to add.
    """
    name = provider.provider_name
    if name in self._providers:
        logger.warning(f"Replacing existing provider: {name}")

    self._providers[name] = provider

    # Register all tools from this provider
    for tool in provider.tools():
        self._routing_table[tool.metadata.name] = (provider, tool)
        self._tools.append(tool)

remove_provider

remove_provider(name: str) -> bool

Remove a provider by name.

Parameters:

Name	Type	Description	Default
name	str	Provider name to remove.	required

Returns:

Type	Description
bool	True if removed, False if not found.

Source code in redis_agent_kit/tools/manager.py

def remove_provider(self, name: str) -> bool:
    """Remove a provider by name.

    Args:
        name: Provider name to remove.

    Returns:
        True if removed, False if not found.
    """
    if name not in self._providers:
        return False

    self._providers.pop(name)

    # Remove all tools from this provider
    tools_to_remove = [t for t in self._tools if t.metadata.provider_name == name]
    for tool in tools_to_remove:
        self._routing_table.pop(tool.metadata.name, None)
        self._tools.remove(tool)

    return True

get_tools

get_tools(capabilities: set[ToolCapability] | None = None, exclude_capabilities: set[ToolCapability] | None = None) -> list[Tool]

Get tools, optionally filtered by capability.

Parameters:

Name	Type	Description	Default
capabilities	set[ToolCapability] | None	If set, only include tools with these capabilities.	None
exclude_capabilities	set[ToolCapability] | None	If set, exclude tools with these capabilities.	None

Returns:

Type	Description
list[Tool]	List of Tool objects.

Source code in redis_agent_kit/tools/manager.py

def get_tools(
    self,
    capabilities: set[ToolCapability] | None = None,
    exclude_capabilities: set[ToolCapability] | None = None,
) -> list[Tool]:
    """Get tools, optionally filtered by capability.

    Args:
        capabilities: If set, only include tools with these capabilities.
        exclude_capabilities: If set, exclude tools with these capabilities.

    Returns:
        List of Tool objects.
    """
    result = self._tools

    if capabilities:
        result = [t for t in result if t.metadata.capability in capabilities]

    if exclude_capabilities:
        result = [t for t in result if t.metadata.capability not in exclude_capabilities]

    return result

get_openai_tools

get_openai_tools(capabilities: set[ToolCapability] | None = None, exclude_capabilities: set[ToolCapability] | None = None) -> list[dict[str, Any]]

Get tools in OpenAI function calling format.

Parameters:

Name	Type	Description	Default
capabilities	set[ToolCapability] | None	If set, only include tools with these capabilities.	None
exclude_capabilities	set[ToolCapability] | None	If set, exclude tools with these capabilities.	None

Returns:

Type	Description
list[dict[str, Any]]	List of OpenAI tool schemas.

Source code in redis_agent_kit/tools/manager.py

def get_openai_tools(
    self,
    capabilities: set[ToolCapability] | None = None,
    exclude_capabilities: set[ToolCapability] | None = None,
) -> list[dict[str, Any]]:
    """Get tools in OpenAI function calling format.

    Args:
        capabilities: If set, only include tools with these capabilities.
        exclude_capabilities: If set, exclude tools with these capabilities.

    Returns:
        List of OpenAI tool schemas.
    """
    tools = self.get_tools(capabilities, exclude_capabilities)
    return [t.to_openai_schema() for t in tools]

execute async

execute(tool_name: str, args: dict[str, Any]) -> Any

Execute a tool by name.

Parameters:

Name	Type	Description	Default
tool_name	str	Name of the tool to execute.	required
args	dict[str, Any]	Arguments to pass to the tool.	required

Returns:

Type	Description
Any	Tool execution result.

Raises:

Type	Description
KeyError	If tool not found.

Source code in redis_agent_kit/tools/manager.py

async def execute(self, tool_name: str, args: dict[str, Any]) -> Any:
    """Execute a tool by name.

    Args:
        tool_name: Name of the tool to execute.
        args: Arguments to pass to the tool.

    Returns:
        Tool execution result.

    Raises:
        KeyError: If tool not found.
    """
    if tool_name not in self._routing_table:
        raise KeyError(f"Tool not found: {tool_name}")

    _, tool = self._routing_table[tool_name]
    return await tool.invoke(args)

has_tool

has_tool(tool_name: str) -> bool

Check if a tool exists.

Source code in redis_agent_kit/tools/manager.py

def has_tool(self, tool_name: str) -> bool:
    """Check if a tool exists."""
    return tool_name in self._routing_table

ToolProvider

ToolProvider

Bases: ABC

Base class for tool providers.

Tool providers expose tools that LLMs can call. Each provider can expose multiple tools and is responsible for implementing them.

Example

class WeatherProvider(ToolProvider): @property def provider_name(self) -> str: return "weather"

@property
def capabilities(self) -> set[ToolCapability]:
    return {ToolCapability.UTILITIES}

def create_tool_schemas(self) -> list[ToolDefinition]:
    return [
        ToolDefinition(
            name="get_weather",
            description="Get current weather for a location",
            capability=ToolCapability.UTILITIES,
            parameters={
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "City name"}
                },
                "required": ["location"],
            },
        )
    ]

async def get_weather(self, location: str) -> dict:
    return {"location": location, "temp": 72, "conditions": "sunny"}

provider_name abstractmethod property

provider_name: str

Unique provider name (e.g., 'weather', 'github').

create_tool_schemas abstractmethod

create_tool_schemas() -> list[ToolDefinition]

Create tool definitions for this provider.

Source code in redis_agent_kit/tools/provider.py

@abstractmethod
def create_tool_schemas(self) -> list[ToolDefinition]:
    """Create tool definitions for this provider."""
    ...

tools

tools() -> list[Tool]

Return Tool objects ready for use.

Builds Tool objects from create_tool_schemas(), wiring each to the corresponding method on this provider.

Source code in redis_agent_kit/tools/provider.py

def tools(self) -> list[Tool]:
    """Return Tool objects ready for use.

    Builds Tool objects from create_tool_schemas(), wiring each to the
    corresponding method on this provider.
    """
    schemas = self.create_tool_schemas()
    tools: list[Tool] = []

    for schema in schemas:
        meta = ToolMetadata(
            name=schema.name,
            description=schema.description,
            capability=schema.capability,
            provider_name=self.provider_name,
        )

        # Find the method for this tool
        op_name = self._resolve_operation(schema.name)
        method = getattr(self, op_name, None) if op_name else None

        if not callable(method):
            raise RuntimeError(
                f"Provider {self.__class__.__name__} has no method {op_name!r} "
                f"for tool {schema.name!r}."
            )

        # Capture method via default arg
        async def _invoke(args: dict[str, Any], _method: Any = method) -> Any:
            return await _method(**(args or {}))

        tools.append(Tool(metadata=meta, definition=schema, invoke=_invoke))

    return tools

ToolCapability

ToolCapability

Bases: StrEnum

Capabilities that tools can provide.

ToolMetadata

ToolMetadata

Bases: BaseModel

Metadata about a tool implementation.