Skip to content

Tool Base Classes

Core base classes and data structures for the tools framework.

Tool

The base class for all tools that provides a common interface for both synchronous and asynchronous tool execution.

rllm.tools.tool_base.Tool

Abstract base class for all tools that provides a common interface.

All tools should inherit from this class and implement either: - forward() for synchronous tools - async_forward() for asynchronous tools

Source code in rllm/tools/tool_base.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
class Tool:
    """
    Abstract base class for all tools that provides a common interface.

    All tools should inherit from this class and implement either:
    - forward() for synchronous tools
    - async_forward() for asynchronous tools
    """

    def __init__(self, name: str | None = None, description: str | None = None, function: Callable | None = None):
        """
        Initialize the tool with a name.

        Args:
            name (str): The name of the tool, used for tool registry and calling.
            description (str): A description of the tool's purpose and functionality.
            function (Callable, optional): Function to convert to tool format.
        """
        self.name = name
        self.description = description
        self.function = function

        if function is not None:
            self._json = function_to_dict(function)
            self.name = self._json["function"]["name"]
            self.description = self._json["function"]["description"]
        else:
            assert name is not None, "Name is required for Tool class."
            assert description is not None, "Description is required for Tool class."
            # User must provide json
            self._json = self.json
            assert self._json is not None, "Json representation of the tool is required."

    @property
    def json(self) -> dict[str, Any]:
        """
        Return the tool's information in a standardized format for tool registration.

        Returns:
            Dict[str, Any]: Tool information including name, description, and parameters.
                Expected format:
                {
                    "type": "function",
                    "function": {
                        "name": self.name,
                        "description": "Description of what the tool does",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                # Parameter definitions
                            },
                            "required": ["list", "of", "required", "parameters"]
                        }
                    }
                }
        """
        return self._json

    def forward(self, *args, **kwargs) -> ToolOutput:
        """
        Synchronous implementation of the tool functionality.
        Override this method for synchronous tools.

        Returns:
            Any: The result of the tool execution.
        """
        if self.function is not None:
            try:
                output = self.function(*args, **kwargs)
                return ToolOutput(name=self.name or "unknown", output=output)
            except Exception as e:
                return ToolOutput(name=self.name or "unknown", error=f"{type(e).__name__} - {str(e)}")
        else:
            raise NotImplementedError("Tool must implement forward() method or provide a function")

    async def async_forward(self, *args, **kwargs) -> ToolOutput:
        """
        Asynchronous implementation of the tool functionality.
        Override this method for asynchronous tools.

        Returns:
            Any: The result of the tool execution.
        """
        return self.forward(*args, **kwargs)

    def __call__(self, *args, use_async=False, **kwargs):
        """
        Make the tool instance callable.
        - If use_async is True, delegates to async_forward
        - If use_async is False, delegates to forward
        - If use_async is None (default), uses async_forward if implemented, otherwise forward

        Args:
            *args: Positional arguments to pass to the implementation
            use_async: Whether to use the async implementation (if None, auto-detect)
            **kwargs: Keyword arguments to pass to the implementation

        Returns:
            Any: The result of the tool execution, which may be a coroutine if using async_forward
        """

        has_async = inspect.isfunction(self.__class__.async_forward) or self.__class__.async_forward is Tool.async_forward
        has_sync = not (inspect.isfunction(self.__class__.forward) and self.__class__.forward is Tool.forward)
        # Explicit routing based on use_async flag
        if use_async is True:
            if has_async:
                return self.async_forward(*args, **kwargs)
            else:
                raise NotImplementedError(f"Tool {self.__class__.__name__} does not implement async_forward() but use_async=True was specified.")
        elif use_async is False:
            return self.forward(*args, **kwargs)

        # Auto-detect implementation if use_async is None
        if has_async:
            return self.async_forward(*args, **kwargs)
        elif has_sync:
            return self.forward(*args, **kwargs)
        else:
            raise NotImplementedError(f"Tool {self.__class__.__name__} must implement either forward() or async_forward().")

json property 

json: dict[str, Any]

Return the tool's information in a standardized format for tool registration.

Returns:

Type Description
dict[str, Any]

Dict[str, Any]: Tool information including name, description, and parameters. Expected format: { "type": "function", "function": { "name": self.name, "description": "Description of what the tool does", "parameters": { "type": "object", "properties": { # Parameter definitions }, "required": ["list", "of", "required", "parameters"] } } }

__init__ 

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

Initialize the tool with a name.

Parameters:

Name Type Description Default
name str

The name of the tool, used for tool registry and calling.

 None
description str

A description of the tool's purpose and functionality.

 None
function Callable

Function to convert to tool format.

 None
Source code in rllm/tools/tool_base.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def __init__(self, name: str | None = None, description: str | None = None, function: Callable | None = None):
    """
    Initialize the tool with a name.

    Args:
        name (str): The name of the tool, used for tool registry and calling.
        description (str): A description of the tool's purpose and functionality.
        function (Callable, optional): Function to convert to tool format.
    """
    self.name = name
    self.description = description
    self.function = function

    if function is not None:
        self._json = function_to_dict(function)
        self.name = self._json["function"]["name"]
        self.description = self._json["function"]["description"]
    else:
        assert name is not None, "Name is required for Tool class."
        assert description is not None, "Description is required for Tool class."
        # User must provide json
        self._json = self.json
        assert self._json is not None, "Json representation of the tool is required."

forward 

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

Synchronous implementation of the tool functionality. Override this method for synchronous tools.

Returns:

Name Type Description
Any ToolOutput

The result of the tool execution.
Source code in rllm/tools/tool_base.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def forward(self, *args, **kwargs) -> ToolOutput:
    """
    Synchronous implementation of the tool functionality.
    Override this method for synchronous tools.

    Returns:
        Any: The result of the tool execution.
    """
    if self.function is not None:
        try:
            output = self.function(*args, **kwargs)
            return ToolOutput(name=self.name or "unknown", output=output)
        except Exception as e:
            return ToolOutput(name=self.name or "unknown", error=f"{type(e).__name__} - {str(e)}")
    else:
        raise NotImplementedError("Tool must implement forward() method or provide a function")

async_forward async 

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

Asynchronous implementation of the tool functionality. Override this method for asynchronous tools.

Returns:

Name Type Description
Any ToolOutput

The result of the tool execution.
Source code in rllm/tools/tool_base.py 
124
125
126
127
128
129
130
131
132
async def async_forward(self, *args, **kwargs) -> ToolOutput:
    """
    Asynchronous implementation of the tool functionality.
    Override this method for asynchronous tools.

    Returns:
        Any: The result of the tool execution.
    """
    return self.forward(*args, **kwargs)

__call__ 

__call__(*args, use_async=False, **kwargs)

Make the tool instance callable. - If use_async is True, delegates to async_forward - If use_async is False, delegates to forward - If use_async is None (default), uses async_forward if implemented, otherwise forward

Parameters:

Name Type Description Default
*args

Positional arguments to pass to the implementation

 ()
use_async

Whether to use the async implementation (if None, auto-detect)

 False
**kwargs

Keyword arguments to pass to the implementation

 {}

Returns:

Name Type Description
Any

The result of the tool execution, which may be a coroutine if using async_forward
Source code in rllm/tools/tool_base.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def __call__(self, *args, use_async=False, **kwargs):
    """
    Make the tool instance callable.
    - If use_async is True, delegates to async_forward
    - If use_async is False, delegates to forward
    - If use_async is None (default), uses async_forward if implemented, otherwise forward

    Args:
        *args: Positional arguments to pass to the implementation
        use_async: Whether to use the async implementation (if None, auto-detect)
        **kwargs: Keyword arguments to pass to the implementation

    Returns:
        Any: The result of the tool execution, which may be a coroutine if using async_forward
    """

    has_async = inspect.isfunction(self.__class__.async_forward) or self.__class__.async_forward is Tool.async_forward
    has_sync = not (inspect.isfunction(self.__class__.forward) and self.__class__.forward is Tool.forward)
    # Explicit routing based on use_async flag
    if use_async is True:
        if has_async:
            return self.async_forward(*args, **kwargs)
        else:
            raise NotImplementedError(f"Tool {self.__class__.__name__} does not implement async_forward() but use_async=True was specified.")
    elif use_async is False:
        return self.forward(*args, **kwargs)

    # Auto-detect implementation if use_async is None
    if has_async:
        return self.async_forward(*args, **kwargs)
    elif has_sync:
        return self.forward(*args, **kwargs)
    else:
        raise NotImplementedError(f"Tool {self.__class__.__name__} must implement either forward() or async_forward().")

ToolCall

Data structure representing a tool call with name and arguments.

rllm.tools.tool_base.ToolCall dataclass

Source code in rllm/tools/tool_base.py 
 9
10
11
12
13
14
15
@dataclass
class ToolCall:
    name: str
    arguments: dict[str, Any]

    def to_dict(self):
        return {"name": self.name, "arguments": self.arguments}

ToolOutput

Data structure representing the output of a tool execution, including results, errors, and metadata.

rllm.tools.tool_base.ToolOutput dataclass

Source code in rllm/tools/tool_base.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@dataclass
class ToolOutput:
    name: str
    output: str | list | dict | None = None
    error: str | None = None
    metadata: dict | None = None

    def __str__(self) -> str:
        """Convert the tool output to a string representation."""
        if self.error:
            return f"Error: {self.error}"
        elif self.output is None:
            return ""
        elif isinstance(self.output, list | dict):
            import json

            return json.dumps(self.output)
        else:
            return str(self.output)

    def to_string(self) -> str:
        """
        Convert the tool output to a string representation.

        Example usage:
            tool_output = ToolOutput(name="calculator", output=42)
            result_string = tool_output.to_string()
        """
        return str(self)

__str__ 

__str__() -> str

Convert the tool output to a string representation.

Source code in rllm/tools/tool_base.py 
25
26
27
28
29
30
31
32
33
34
35
36
def __str__(self) -> str:
    """Convert the tool output to a string representation."""
    if self.error:
        return f"Error: {self.error}"
    elif self.output is None:
        return ""
    elif isinstance(self.output, list | dict):
        import json

        return json.dumps(self.output)
    else:
        return str(self.output)

to_string 

to_string() -> str

Convert the tool output to a string representation.

Example usage

tool_output = ToolOutput(name="calculator", output=42) result_string = tool_output.to_string()

Source code in rllm/tools/tool_base.py 
38
39
40
41
42
43
44
45
46
def to_string(self) -> str:
    """
    Convert the tool output to a string representation.

    Example usage:
        tool_output = ToolOutput(name="calculator", output=42)
        result_string = tool_output.to_string()
    """
    return str(self)