18124993824

Committed 30 Sep 2025 09:17AM UTC coverage: 91.987% (-0.07%) from 92.055%

Build # 18124993824

Build Type

Pull #9849

github

Committed by

web-flow

Commit Message

Merge 88d2656a0 into 34aa66ecc

Pull Request Pull Request #9849: Add Tool/Toolset warm_up

Run Details

13237 of 14390 relevant lines covered (91.99%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.72

haystack/tools/toolset.py

# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

from dataclasses import dataclass, field
from typing import Any, Iterator, Union

from haystack.core.serialization import generate_qualified_class_name, import_class_by_name
from haystack.tools.tool import Tool, _check_duplicate_tool_names


@dataclass
class Toolset:
    """
    A collection of related Tools that can be used and managed as a cohesive unit.

    Toolset serves two main purposes:

    1. Group related tools together:
       Toolset allows you to organize related tools into a single collection, making it easier
       to manage and use them as a unit in Haystack pipelines.

       Example:
       ```python
       from haystack.tools import Tool, Toolset
       from haystack.components.tools import ToolInvoker

       # Define math functions
       def add_numbers(a: int, b: int) -> int:
           return a + b

       def subtract_numbers(a: int, b: int) -> int:
           return a - b

       # Create tools with proper schemas
       add_tool = Tool(
           name="add",
           description="Add two numbers",
           parameters={
               "type": "object",
               "properties": {
                   "a": {"type": "integer"},
                   "b": {"type": "integer"}
               },
               "required": ["a", "b"]
           },
           function=add_numbers
       )

       subtract_tool = Tool(
           name="subtract",
           description="Subtract b from a",
           parameters={
               "type": "object",
               "properties": {
                   "a": {"type": "integer"},
                   "b": {"type": "integer"}
               },
               "required": ["a", "b"]
           },
           function=subtract_numbers
       )

       # Create a toolset with the math tools
       math_toolset = Toolset([add_tool, subtract_tool])

       # Use the toolset with a ToolInvoker or ChatGenerator component
       invoker = ToolInvoker(tools=math_toolset)
       ```

    2. Base class for dynamic tool loading:
       By subclassing Toolset, you can create implementations that dynamically load tools
       from external sources like OpenAPI URLs, MCP servers, or other resources.

       Example:
       ```python
       from haystack.core.serialization import generate_qualified_class_name
       from haystack.tools import Tool, Toolset
       from haystack.components.tools import ToolInvoker

       class CalculatorToolset(Toolset):
           '''A toolset for calculator operations.'''

           def __init__(self):
               tools = self._create_tools()
               super().__init__(tools)

           def _create_tools(self):
               # These Tool instances are obviously defined statically and for illustration purposes only.
               # In a real-world scenario, you would dynamically load tools from an external source here.
               tools = []
               add_tool = Tool(
                   name="add",
                   description="Add two numbers",
                   parameters={
                       "type": "object",
                       "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
                       "required": ["a", "b"],
                   },
                   function=lambda a, b: a + b,
               )

               multiply_tool = Tool(
                   name="multiply",
                   description="Multiply two numbers",
                   parameters={
                       "type": "object",
                       "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
                       "required": ["a", "b"],
                   },
                   function=lambda a, b: a * b,
               )

               tools.append(add_tool)
               tools.append(multiply_tool)

               return tools

           def to_dict(self):
               return {
                   "type": generate_qualified_class_name(type(self)),
                   "data": {},  # no data to serialize as we define the tools dynamically
               }

           @classmethod
           def from_dict(cls, data):
               return cls()  # Recreate the tools dynamically during deserialization

       # Create the dynamic toolset and use it with ToolInvoker
       calculator_toolset = CalculatorToolset()
       invoker = ToolInvoker(tools=calculator_toolset)
       ```

    Toolset implements the collection interface (__iter__, __contains__, __len__, __getitem__),
    making it behave like a list of Tools. This makes it compatible with components that expect
    iterable tools, such as ToolInvoker or Haystack chat generators.

    When implementing a custom Toolset subclass for dynamic tool loading:
    - Perform the dynamic loading in the __init__ method
    - Override to_dict() and from_dict() methods if your tools are defined dynamically
    - Serialize endpoint descriptors rather than tool instances if your tools
      are loaded from external sources
    """

    # Use field() with default_factory to initialize the list
    tools: list[Tool] = field(default_factory=list)

    def __post_init__(self):
        """
        Validate and set up the toolset after initialization.

        This handles the case when tools are provided during initialization.
        """
        # If initialization was done a single Tool, raise an error
        if isinstance(self.tools, Tool):
            raise TypeError("A single Tool cannot be directly passed to Toolset. Please use a list: Toolset([tool])")

        # Check for duplicate tool names in the initial set
        _check_duplicate_tool_names(self.tools)

    def __iter__(self) -> Iterator[Tool]:
        """
        Return an iterator over the Tools in this Toolset.

        This allows the Toolset to be used wherever a list of Tools is expected.

        :returns: An iterator yielding Tool instances
        """
        return iter(self.tools)

    def __contains__(self, item: Any) -> bool:
        """
        Check if a tool is in this Toolset.

        Supports checking by:
        - Tool instance: tool in toolset
        - Tool name: "tool_name" in toolset

        :param item: Tool instance or tool name string
        :returns: True if contained, False otherwise
        """
        if isinstance(item, str):
            return any(tool.name == item for tool in self.tools)
        if isinstance(item, Tool):
            return item in self.tools
        return False

    def warm_up(self) -> None:
        """
        Warm up the Toolset.
        """
        pass

    def add(self, tool: Union[Tool, "Toolset"]) -> None:
        """
        Add a new Tool or merge another Toolset.

        :param tool: A Tool instance or another Toolset to add
        :raises ValueError: If adding the tool would result in duplicate tool names
        :raises TypeError: If the provided object is not a Tool or Toolset
        """
        new_tools = []

        if isinstance(tool, Tool):
            new_tools = [tool]
        elif isinstance(tool, Toolset):
            new_tools = list(tool)
        else:
            raise TypeError(f"Expected Tool or Toolset, got {type(tool).__name__}")

        # Check for duplicates before adding
        combined_tools = self.tools + new_tools
        _check_duplicate_tool_names(combined_tools)

        self.tools.extend(new_tools)

    def to_dict(self) -> dict[str, Any]:
        """
        Serialize the Toolset to a dictionary.

        :returns: A dictionary representation of the Toolset

        Note for subclass implementers:
        The default implementation is ideal for scenarios where Tool resolution is static. However, if your subclass
        of Toolset dynamically resolves Tool instances from external sources—such as an MCP server, OpenAPI URL, or
        a local OpenAPI specification—you should consider serializing the endpoint descriptor instead of the Tool
        instances themselves. This strategy preserves the dynamic nature of your Toolset and minimizes the overhead
        associated with serializing potentially large collections of Tool objects. Moreover, by serializing the
        descriptor, you ensure that the deserialization process can accurately reconstruct the Tool instances, even
        if they have been modified or removed since the last serialization. Failing to serialize the descriptor may
        lead to issues where outdated or incorrect Tool configurations are loaded, potentially causing errors or
        unexpected behavior.
        """
        return {
            "type": generate_qualified_class_name(type(self)),
            "data": {"tools": [tool.to_dict() for tool in self.tools]},
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "Toolset":
        """
        Deserialize a Toolset from a dictionary.

        :param data: Dictionary representation of the Toolset
        :returns: A new Toolset instance
        """
        inner_data = data["data"]
        tools_data = inner_data.get("tools", [])

        tools = []
        for tool_data in tools_data:
            tool_class = import_class_by_name(tool_data["type"])
            if not issubclass(tool_class, Tool):
                raise TypeError(f"Class '{tool_class}' is not a subclass of Tool")
            tools.append(tool_class.from_dict(tool_data))

        return cls(tools=tools)

    def __add__(self, other: Union[Tool, "Toolset", list[Tool]]) -> "Toolset":
        """
        Concatenate this Toolset with another Tool, Toolset, or list of Tools.

        :param other: Another Tool, Toolset, or list of Tools to concatenate
        :returns: A new Toolset containing all tools
        :raises TypeError: If the other parameter is not a Tool, Toolset, or list of Tools
        :raises ValueError: If the combination would result in duplicate tool names
        """
        if isinstance(other, Tool):
            combined_tools = self.tools + [other]
        elif isinstance(other, Toolset):
            return _ToolsetWrapper([self, other])
        elif isinstance(other, list) and all(isinstance(item, Tool) for item in other):
            combined_tools = self.tools + other
        else:
            raise TypeError(f"Cannot add {type(other).__name__} to Toolset")

        # Check for duplicates
        _check_duplicate_tool_names(combined_tools)

        return Toolset(tools=combined_tools)

    def __len__(self) -> int:
        """
        Return the number of Tools in this Toolset.

        :returns: Number of Tools
        """
        return len(self.tools)

    def __getitem__(self, index):
        """
        Get a Tool by index.

        :param index: Index of the Tool to get
        :returns: The Tool at the specified index
        """
        return self.tools[index]


class _ToolsetWrapper(Toolset):
    """
    A wrapper that holds multiple toolsets and provides a unified interface.

    This is used internally when combining different types of toolsets to preserve
    their individual configurations while still being usable with ToolInvoker.
    """

    def __init__(self, toolsets: list[Toolset]):
        self.toolsets = toolsets
        # Check for duplicate tool names across all toolsets
        all_tools = []
        for toolset in toolsets:
            all_tools.extend(list(toolset))
        _check_duplicate_tool_names(all_tools)
        super().__init__(tools=all_tools)

    def __iter__(self):
        """Iterate over all tools from all toolsets."""
        for toolset in self.toolsets:
            yield from toolset

    def __contains__(self, item):
        """Check if a tool is in any of the toolsets."""
        return any(item in toolset for toolset in self.toolsets)

    def warm_up(self):
        """Warm up all toolsets."""
        for toolset in self.toolsets:
            toolset.warm_up()

    def __len__(self):
        """Return total number of tools across all toolsets."""
        return sum(len(toolset) for toolset in self.toolsets)

    def __getitem__(self, index):
        """Get a tool by index across all toolsets."""
        current_index = 0
        for toolset in self.toolsets:
            toolset_len = len(toolset)
            if current_index + toolset_len > index:
                return toolset[index - current_index]
            current_index += toolset_len
        raise IndexError("ToolsetWrapper index out of range")

    def __add__(self, other):
        """Add another toolset or tool to this wrapper."""
        # Import here to avoid circular reference issues
        from haystack.tools.toolset import Toolset

        if isinstance(other, Toolset):
            # Add the toolset to our list
            new_toolsets = self.toolsets + [other]
        elif isinstance(other, Tool):
            # Convert tool to a basic toolset and add it
            new_toolsets = self.toolsets + [Toolset([other])]
        elif isinstance(other, list) and all(isinstance(item, Tool) for item in other):
            # Convert list of tools to a basic toolset and add it
            new_toolsets = self.toolsets + [Toolset(other)]
        else:
            raise TypeError(f"Cannot add {type(other).__name__} to ToolsetWrapper")

        return _ToolsetWrapper(new_toolsets)

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	from dataclasses import dataclass, field	1✔
6	from typing import Any, Iterator, Union	1✔
7
8	from haystack.core.serialization import generate_qualified_class_name, import_class_by_name	1✔
9	from haystack.tools.tool import Tool, _check_duplicate_tool_names	1✔
10
11
12	@dataclass	1✔
13	class Toolset:	1✔
14	"""
15	A collection of related Tools that can be used and managed as a cohesive unit.
16
17	Toolset serves two main purposes:
18
19	1. Group related tools together:
20	Toolset allows you to organize related tools into a single collection, making it easier
21	to manage and use them as a unit in Haystack pipelines.
22
23	Example:
24	```python
25	from haystack.tools import Tool, Toolset
26	from haystack.components.tools import ToolInvoker
27
28	# Define math functions
29	def add_numbers(a: int, b: int) -> int:
30	return a + b
31
32	def subtract_numbers(a: int, b: int) -> int:
33	return a - b
34
35	# Create tools with proper schemas
36	add_tool = Tool(
37	name="add",
38	description="Add two numbers",
39	parameters={
40	"type": "object",
41	"properties": {
42	"a": {"type": "integer"},
43	"b": {"type": "integer"}
44	},
45	"required": ["a", "b"]
46	},
47	function=add_numbers
48	)
49
50	subtract_tool = Tool(
51	name="subtract",
52	description="Subtract b from a",
53	parameters={
54	"type": "object",
55	"properties": {
56	"a": {"type": "integer"},
57	"b": {"type": "integer"}
58	},
59	"required": ["a", "b"]
60	},
61	function=subtract_numbers
62	)
63
64	# Create a toolset with the math tools
65	math_toolset = Toolset([add_tool, subtract_tool])
66
67	# Use the toolset with a ToolInvoker or ChatGenerator component
68	invoker = ToolInvoker(tools=math_toolset)
69	```
70
71	2. Base class for dynamic tool loading:
72	By subclassing Toolset, you can create implementations that dynamically load tools
73	from external sources like OpenAPI URLs, MCP servers, or other resources.
74
75	Example:
76	```python
77	from haystack.core.serialization import generate_qualified_class_name
78	from haystack.tools import Tool, Toolset
79	from haystack.components.tools import ToolInvoker
80
81	class CalculatorToolset(Toolset):
82	'''A toolset for calculator operations.'''
83
84	def __init__(self):
85	tools = self._create_tools()
86	super().__init__(tools)
87
88	def _create_tools(self):
89	# These Tool instances are obviously defined statically and for illustration purposes only.
90	# In a real-world scenario, you would dynamically load tools from an external source here.
91	tools = []
92	add_tool = Tool(
93	name="add",
94	description="Add two numbers",
95	parameters={
96	"type": "object",
97	"properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
98	"required": ["a", "b"],
99	},
100	function=lambda a, b: a + b,
101	)
102
103	multiply_tool = Tool(
104	name="multiply",
105	description="Multiply two numbers",
106	parameters={
107	"type": "object",
108	"properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
109	"required": ["a", "b"],
110	},
111	function=lambda a, b: a * b,
112	)
113
114	tools.append(add_tool)
115	tools.append(multiply_tool)
116
117	return tools
118
119	def to_dict(self):
120	return {
121	"type": generate_qualified_class_name(type(self)),
122	"data": {}, # no data to serialize as we define the tools dynamically
123	}
124
125	@classmethod
126	def from_dict(cls, data):
127	return cls() # Recreate the tools dynamically during deserialization
128
129	# Create the dynamic toolset and use it with ToolInvoker
130	calculator_toolset = CalculatorToolset()
131	invoker = ToolInvoker(tools=calculator_toolset)
132	```
133
134	Toolset implements the collection interface (__iter__, __contains__, __len__, __getitem__),
135	making it behave like a list of Tools. This makes it compatible with components that expect
136	iterable tools, such as ToolInvoker or Haystack chat generators.
137
138	When implementing a custom Toolset subclass for dynamic tool loading:
139	- Perform the dynamic loading in the __init__ method
140	- Override to_dict() and from_dict() methods if your tools are defined dynamically
141	- Serialize endpoint descriptors rather than tool instances if your tools
142	are loaded from external sources
143	"""
144
145	# Use field() with default_factory to initialize the list
146	tools: list[Tool] = field(default_factory=list)	1✔
147
148	def __post_init__(self):	1✔
149	"""
150	Validate and set up the toolset after initialization.
151
152	This handles the case when tools are provided during initialization.
153	"""
154	# If initialization was done a single Tool, raise an error
155	if isinstance(self.tools, Tool):	1✔
156	raise TypeError("A single Tool cannot be directly passed to Toolset. Please use a list: Toolset([tool])")	×
157
158	# Check for duplicate tool names in the initial set
159	_check_duplicate_tool_names(self.tools)	1✔
160
161	def __iter__(self) -> Iterator[Tool]:	1✔
162	"""
163	Return an iterator over the Tools in this Toolset.
164
165	This allows the Toolset to be used wherever a list of Tools is expected.
166
167	:returns: An iterator yielding Tool instances
168	"""
169	return iter(self.tools)	1✔
170
171	def __contains__(self, item: Any) -> bool:	1✔
172	"""
173	Check if a tool is in this Toolset.
174
175	Supports checking by:
176	- Tool instance: tool in toolset
177	- Tool name: "tool_name" in toolset
178
179	:param item: Tool instance or tool name string
180	:returns: True if contained, False otherwise
181	"""
182	if isinstance(item, str):	1✔
183	return any(tool.name == item for tool in self.tools)	1✔
184	if isinstance(item, Tool):	1✔
185	return item in self.tools	1✔
186	return False	×
187
188	def warm_up(self) -> None:	1✔
189	"""
190	Warm up the Toolset.
191	"""
192	pass	×
193
194	def add(self, tool: Union[Tool, "Toolset"]) -> None:	1✔
195	"""
196	Add a new Tool or merge another Toolset.
197
198	:param tool: A Tool instance or another Toolset to add
199	:raises ValueError: If adding the tool would result in duplicate tool names
200	:raises TypeError: If the provided object is not a Tool or Toolset
201	"""
202	new_tools = []	1✔
203
204	if isinstance(tool, Tool):	1✔
205	new_tools = [tool]	1✔
206	elif isinstance(tool, Toolset):	×
207	new_tools = list(tool)	×
208	else:
209	raise TypeError(f"Expected Tool or Toolset, got {type(tool).__name__}")	×
210
211	# Check for duplicates before adding
212	combined_tools = self.tools + new_tools	1✔
213	_check_duplicate_tool_names(combined_tools)	1✔
214
215	self.tools.extend(new_tools)	1✔
216
217	def to_dict(self) -> dict[str, Any]:	1✔
218	"""
219	Serialize the Toolset to a dictionary.
220
221	:returns: A dictionary representation of the Toolset
222
223	Note for subclass implementers:
224	The default implementation is ideal for scenarios where Tool resolution is static. However, if your subclass
225	of Toolset dynamically resolves Tool instances from external sources—such as an MCP server, OpenAPI URL, or
226	a local OpenAPI specification—you should consider serializing the endpoint descriptor instead of the Tool
227	instances themselves. This strategy preserves the dynamic nature of your Toolset and minimizes the overhead
228	associated with serializing potentially large collections of Tool objects. Moreover, by serializing the
229	descriptor, you ensure that the deserialization process can accurately reconstruct the Tool instances, even
230	if they have been modified or removed since the last serialization. Failing to serialize the descriptor may
231	lead to issues where outdated or incorrect Tool configurations are loaded, potentially causing errors or
232	unexpected behavior.
233	"""
234	return {	1✔
235	"type": generate_qualified_class_name(type(self)),
236	"data": {"tools": [tool.to_dict() for tool in self.tools]},
237	}
238
239	@classmethod	1✔
240	def from_dict(cls, data: dict[str, Any]) -> "Toolset":	1✔
241	"""
242	Deserialize a Toolset from a dictionary.
243
244	:param data: Dictionary representation of the Toolset
245	:returns: A new Toolset instance
246	"""
247	inner_data = data["data"]	1✔
248	tools_data = inner_data.get("tools", [])	1✔
249
250	tools = []	1✔
251	for tool_data in tools_data:	1✔
252	tool_class = import_class_by_name(tool_data["type"])	1✔
253	if not issubclass(tool_class, Tool):	1✔
254	raise TypeError(f"Class '{tool_class}' is not a subclass of Tool")	×
255	tools.append(tool_class.from_dict(tool_data))	1✔
256
257	return cls(tools=tools)	1✔
258
259	def __add__(self, other: Union[Tool, "Toolset", list[Tool]]) -> "Toolset":	1✔
260	"""
261	Concatenate this Toolset with another Tool, Toolset, or list of Tools.
262
263	:param other: Another Tool, Toolset, or list of Tools to concatenate
264	:returns: A new Toolset containing all tools
265	:raises TypeError: If the other parameter is not a Tool, Toolset, or list of Tools
266	:raises ValueError: If the combination would result in duplicate tool names
267	"""
268	if isinstance(other, Tool):	1✔
269	combined_tools = self.tools + [other]	1✔
270	elif isinstance(other, Toolset):	1✔
271	return _ToolsetWrapper([self, other])	1✔
272	elif isinstance(other, list) and all(isinstance(item, Tool) for item in other):	1✔
273	combined_tools = self.tools + other	1✔
274	else:
275	raise TypeError(f"Cannot add {type(other).__name__} to Toolset")	1✔
276
277	# Check for duplicates
278	_check_duplicate_tool_names(combined_tools)	1✔
279
280	return Toolset(tools=combined_tools)	1✔
281
282	def __len__(self) -> int:	1✔
283	"""
284	Return the number of Tools in this Toolset.
285
286	:returns: Number of Tools
287	"""
288	return len(self.tools)	1✔
289
290	def __getitem__(self, index):	1✔
291	"""
292	Get a Tool by index.
293
294	:param index: Index of the Tool to get
295	:returns: The Tool at the specified index
296	"""
297	return self.tools[index]	1✔
298
299
300	class _ToolsetWrapper(Toolset):	1✔
301	"""
302	A wrapper that holds multiple toolsets and provides a unified interface.
303
304	This is used internally when combining different types of toolsets to preserve
305	their individual configurations while still being usable with ToolInvoker.
306	"""
307
308	def __init__(self, toolsets: list[Toolset]):	1✔
309	self.toolsets = toolsets	1✔
310	# Check for duplicate tool names across all toolsets
311	all_tools = []	1✔
312	for toolset in toolsets:	1✔
313	all_tools.extend(list(toolset))	1✔
314	_check_duplicate_tool_names(all_tools)	1✔
315	super().__init__(tools=all_tools)	1✔
316
317	def __iter__(self):	1✔
318	"""Iterate over all tools from all toolsets."""
319	for toolset in self.toolsets:	1✔
320	yield from toolset	1✔
321
322	def __contains__(self, item):	1✔
323	"""Check if a tool is in any of the toolsets."""
324	return any(item in toolset for toolset in self.toolsets)	1✔
325
326	def warm_up(self):	1✔
327	"""Warm up all toolsets."""
328	for toolset in self.toolsets:	×
329	toolset.warm_up()	×
330
331	def __len__(self):	1✔
332	"""Return total number of tools across all toolsets."""
333	return sum(len(toolset) for toolset in self.toolsets)	1✔
334
335	def __getitem__(self, index):	1✔
336	"""Get a tool by index across all toolsets."""
337	current_index = 0	×
338	for toolset in self.toolsets:	×
339	toolset_len = len(toolset)	×
340	if current_index + toolset_len > index:	×
341	return toolset[index - current_index]	×
342	current_index += toolset_len	×
343	raise IndexError("ToolsetWrapper index out of range")	×
344
345	def __add__(self, other):	1✔
346	"""Add another toolset or tool to this wrapper."""
347	# Import here to avoid circular reference issues
348	from haystack.tools.toolset import Toolset	1✔
349
350	if isinstance(other, Toolset):	1✔
351	# Add the toolset to our list
352	new_toolsets = self.toolsets + [other]	×
353	elif isinstance(other, Tool):	1✔
354	# Convert tool to a basic toolset and add it
355	new_toolsets = self.toolsets + [Toolset([other])]	1✔
356	elif isinstance(other, list) and all(isinstance(item, Tool) for item in other):	×
357	# Convert list of tools to a basic toolset and add it
358	new_toolsets = self.toolsets + [Toolset(other)]	×
359	else:
360	raise TypeError(f"Cannot add {type(other).__name__} to ToolsetWrapper")	×
361
362	return _ToolsetWrapper(new_toolsets)	1✔

deepset-ai / haystack / 18124993824

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous