Tools

Tool system for enabling agent capabilities

The rllm.tools module exposes the data types (Tool, ToolCall, ToolOutput) the rLLM parser uses when reading tool calls out of model responses, plus a small registry. The higher-level MultiTool / ToolEnvironment wrappers that the old Workflow path used have been removed.

For a modern tool-using agent, the recommended pattern is to declare tools as plain Python callables and pass an OpenAI-spec tools list directly to the chat-completions API — see cookbooks/finqa/finqa_tools.py for a worked example with four tools, native OpenAI function calling, and a process-wide SQLite store. Use the data types below if you need to parse tool calls out of a raw text response (e.g. for non-native tool-call formats); see Parsers.

Tool

Base class for all tools.

from rllm.tools import Tool

Constructor

def __init__(
    name: str | None = None,
    description: str | None = None,
    function: Callable | None = None
)

name · str | None

Name of the tool. Required if function is not provided.

description · str | None

Description of the tool’s purpose. Required if function is not provided.

function · Callable | None

Function to convert to tool format. If provided, name and description are auto-extracted from docstring.

Properties

json · dict

Tool schema in OpenAI function calling format.

Methods

forward

Synchronous implementation of tool functionality.

def forward(*args, **kwargs) -> ToolOutput

output · ToolOutput

Tool execution result.

async_forward

Asynchronous implementation of tool functionality.

async def async_forward(*args, **kwargs) -> ToolOutput

call

Make the tool callable.

result = tool(arg1, arg2, use_async=False)

use_async · bool | None

Whether to use async implementation. Auto-detects if None.

---

ToolOutput

Dataclass for tool execution results.

from rllm.tools import ToolOutput

output = ToolOutput(
    name="search",
    output={"results": [...]},
    error=None,
    metadata={"query_time": 0.5}
)

Fields

name · str

Name of the tool that produced this output.

output · str | list | dict | None

The tool’s output data.

error · str | None

Error message if execution failed.

metadata · dict | None

Additional metadata about the execution.

Methods

to_string

Convert output to string representation.

result_str = output.to_string()

---

ToolCall

Dataclass representing a tool call.

from rllm.tools import ToolCall

call = ToolCall(
    name="calculator",
    arguments={"a": 5, "b": 3, "operation": "add"}
)

Fields

name · str

Name of the tool to call.

arguments · dict[str, Any]

Arguments to pass to the tool.

---

ToolRegistry

Singleton registry for managing tools.

from rllm.tools import ToolRegistry

registry = ToolRegistry()

Methods

register

Register a tool class.

registry.register("search", SearchTool)

name · str

Name to register the tool under.

tool_cls · type[Tool]

Tool class to register.

register_all

Register multiple tools at once.

registry.register_all({
    "search": SearchTool,
    "calculator": CalculatorTool
})

get

Get a tool class by name.

tool_cls = registry.get("search")

tool_cls · type[Tool] | None

Tool class if found, None otherwise.

instantiate

Instantiate a tool by name.

tool = registry.instantiate("search", api_key="...")

name · str

Name of the tool to instantiate.

*args, **kwargs · Any

Arguments to pass to tool constructor.

list_tools

List all registered tool names.

tool_names = registry.list_tools()

clear

Clear all registered tools.

registry.clear()

unregister

Unregister a tool by name.

success = registry.unregister("search")

---

Example: Creating a Tool from Function

from rllm.tools import Tool

def add(a: int, b: int) -> int:
    """Adds two numbers.
    
    Args:
        a (int): The first number to be added.
        b (int): The second number to be added.
    
    Returns:
        int: The sum of the two numbers.
    """
    return a + b

# Auto-generate tool from function
add_tool = Tool(function=add)

print(add_tool.json)  # OpenAI function schema
print(add_tool(5, 3))  # ToolOutput(name="add", output=8)

---

Example: Creating a Custom Tool Class

from rllm.tools import Tool, ToolOutput
import requests

class WeatherTool(Tool):
    def __init__(self, api_key: str):
        self.api_key = api_key
        super().__init__(
            name="get_weather",
            description="Get current weather for a location"
        )
    
    @property
    def json(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {
                            "type": "string",
                            "description": "City name"
                        }
                    },
                    "required": ["location"]
                }
            }
        }
    
    def forward(self, location: str) -> ToolOutput:
        try:
            # Call weather API
            response = requests.get(
                f"https://api.weather.com/v1/current",
                params={"location": location, "key": self.api_key}
            )
            data = response.json()
            
            return ToolOutput(
                name=self.name,
                output=data,
                metadata={"location": location}
            )
        except Exception as e:
            return ToolOutput(
                name=self.name,
                error=str(e)
            )

# Use the tool
tool = WeatherTool(api_key="your_key")
result = tool(location="San Francisco")
print(result.output)

---

Example: Async Tool

from rllm.tools import Tool, ToolOutput
import httpx

class AsyncSearchTool(Tool):
    def __init__(self, api_key: str):
        self.api_key = api_key
        super().__init__(
            name="search",
            description="Search the web for information"
        )
    
    @property
    def json(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "Search query"
                        }
                    },
                    "required": ["query"]
                }
            }
        }
    
    async def async_forward(self, query: str) -> ToolOutput:
        async with httpx.AsyncClient() as client:
            try:
                response = await client.get(
                    "https://api.search.com/v1/search",
                    params={"q": query, "key": self.api_key}
                )
                results = response.json()
                
                return ToolOutput(
                    name=self.name,
                    output=results,
                    metadata={"query": query}
                )
            except Exception as e:
                return ToolOutput(
                    name=self.name,
                    error=str(e)
                )

# Use async tool
import asyncio

tool = AsyncSearchTool(api_key="your_key")
result = asyncio.run(tool(query="rLLM framework", use_async=True))
print(result.output)

---

Example: Tool Registry

from rllm.tools import Tool, ToolRegistry, ToolOutput

class CalculatorTool(Tool):
    def __init__(self):
        super().__init__(
            name="calculator",
            description="Perform arithmetic operations"
        )
    
    @property
    def json(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": {
                        "a": {"type": "number"},
                        "b": {"type": "number"},
                        "operation": {
                            "type": "string",
                            "enum": ["add", "subtract", "multiply", "divide"]
                        }
                    },
                    "required": ["a", "b", "operation"]
                }
            }
        }
    
    def forward(self, a: float, b: float, operation: str) -> ToolOutput:
        ops = {
            "add": a + b,
            "subtract": a - b,
            "multiply": a * b,
            "divide": a / b if b != 0 else None
        }
        result = ops.get(operation)
        
        if result is None:
            return ToolOutput(name=self.name, error="Invalid operation or division by zero")
        
        return ToolOutput(name=self.name, output=result)

# Register tool
registry = ToolRegistry()
registry.register("calculator", CalculatorTool)

# Instantiate and use
calc = registry.instantiate("calculator")
result = calc(a=10, b=5, operation="multiply")
print(result.output)  # 50

# List all tools
print(registry.list_tools())  # ["calculator"]