Coverage for haystack/tools/toolset.py: 93%
107 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5import warnings
6from collections.abc import Iterator
7from dataclasses import dataclass, field
8from typing import Any
10from haystack.core.serialization import generate_qualified_class_name, import_class_by_name
11from haystack.tools.tool import Tool, _check_duplicate_tool_names
14@dataclass
15class Toolset:
16 """
17 A collection of related Tools that can be used and managed as a cohesive unit.
19 Toolset serves two main purposes:
21 1. Group related tools together:
22 Toolset allows you to organize related tools into a single collection, making it easier
23 to manage and use them as a unit in Haystack pipelines.
25 Example:
26 ```python
27 from typing import Annotated
28 from haystack.tools import tool, Toolset
29 from haystack.components.agents import Agent
30 from haystack.components.generators.chat import OpenAIChatGenerator
32 # Create tools with the @tool decorator (the recommended way)
33 @tool
34 def add(a: Annotated[int, "first number"], b: Annotated[int, "second number"]) -> int:
35 '''Add two numbers.'''
36 return a + b
38 @tool
39 def subtract(a: Annotated[int, "first number"], b: Annotated[int, "second number"]) -> int:
40 '''Subtract b from a.'''
41 return a - b
43 # Create a toolset with the math tools
44 math_toolset = Toolset([add, subtract])
46 # Use the toolset with an Agent
47 agent = Agent(chat_generator=OpenAIChatGenerator(), tools=math_toolset)
48 ```
50 2. Base class for dynamic tool loading:
51 By subclassing Toolset, you can create implementations that dynamically load tools from external sources like
52 OpenAPI URLs, MCP servers, or other resources.
54 When implementing a custom Toolset subclass for dynamic tool loading:
55 - Load the tools in `warm_up()` and assign them to `self.tools`. Since `warm_up()` may be called before
56 every run, make it idempotent by guarding on your own state (e.g. `if self._client is not None: return`).
57 - Override `to_dict()` and `from_dict()` to serialize the endpoint descriptor (URL, server info) rather than
58 the dynamically loaded Tool instances.
60 Example:
61 ```python
62 from haystack.core.serialization import generate_qualified_class_name
63 from haystack.tools import Toolset
65 class RemoteServiceToolset(Toolset):
66 def __init__(self, endpoint: str) -> None:
67 self.endpoint = endpoint
68 self._client = None
69 super().__init__(tools=[]) # tools are loaded on warm_up()
71 def warm_up(self) -> None:
72 if self._client is not None:
73 return
74 self._client = connect(self.endpoint)
75 self.tools = self._client.fetch_tools()
77 def to_dict(self):
78 return {
79 "type": generate_qualified_class_name(type(self)),
80 "data": {"endpoint": self.endpoint},
81 }
83 @classmethod
84 def from_dict(cls, data):
85 return cls(endpoint=data["data"]["endpoint"])
86 ```
88 Toolset implements the collection interface (__iter__, __contains__, __len__, __getitem__), making it behave like
89 a list of Tools. This makes it compatible with components that expect iterable tools, such as Agent or Haystack
90 chat generators.
91 """
93 # Use field() with default_factory to initialize the list
94 tools: list[Tool] = field(default_factory=list)
96 def __post_init__(self) -> None:
97 """
98 Validate the tools provided during initialization.
99 """
100 # If initialization was done a single Tool, raise an error
101 if isinstance(self.tools, Tool):
102 raise TypeError("A single Tool cannot be directly passed to Toolset. Please use a list: Toolset([tool])")
104 # Check for duplicate tool names in the initial set
105 _check_duplicate_tool_names(self.tools)
107 def __iter__(self) -> Iterator[Tool]:
108 """
109 Return an iterator over the Tools in this Toolset.
111 This allows the Toolset to be used wherever a list of Tools is expected.
113 :returns: An iterator yielding Tool instances
114 """
115 return iter(self.tools)
117 def get_selectable_tools(self) -> list[Tool]:
118 """
119 Return the tools available for name-based selection (e.g. via `Agent.run(tools=["tool_name"])`).
121 Warms up the Toolset first, so lazily loaded tools are selectable too. Subclasses whose iteration does
122 not surface every selectable tool (e.g. SearchableToolset) override this to return the full set.
124 :returns: The list of tools available for name-based selection.
125 """
126 self.warm_up()
127 return list(self.tools)
129 def spawn(self, selected_tool_names: set[str] | None = None) -> "Toolset": # noqa: ARG002
130 """
131 Return this Toolset, or an isolated copy of it, for a single run.
133 A plain Toolset has no run-scoped state, so the default implementation returns `self` and ignores the
134 selection (the Agent materializes it). Subclasses with run-scoped state (e.g. SearchableToolset) override
135 this to return a copy carrying the selection, so concurrent runs sharing the same configured Toolset
136 don't corrupt each other.
138 :param selected_tool_names: Optional tool names this run is restricted to. None means no restriction.
139 :returns: This Toolset, or a run-scoped copy of it.
140 """
141 return self
143 def __contains__(self, item: str | Tool) -> bool:
144 """
145 Check if a tool is in this Toolset.
147 Supports checking by:
148 - Tool instance: tool in toolset
149 - Tool name: "tool_name" in toolset
151 :param item: Tool instance or tool name string
152 :returns: True if contained, False otherwise
153 """
154 if isinstance(item, str):
155 return any(tool.name == item for tool in self)
156 if isinstance(item, Tool):
157 return any(tool is item or tool == item for tool in self)
158 return False
160 def warm_up(self) -> None:
161 """
162 Prepare the Toolset for use.
164 By default, this method iterates through and warms up all tools in the Toolset.
165 Subclasses can override this method to customize initialization behavior, such as:
167 - Setting up shared resources (database connections, HTTP sessions) instead of
168 warming individual tools
169 - Loading tools dynamically from an external source and assigning them to `self.tools`
170 - Controlling when and how tools are initialized
172 For example, a Toolset that manages tools from an external service (like MCPToolset)
173 might override this to initialize a shared connection and load the tools through it:
175 ```python
176 class MCPToolset(Toolset):
177 def warm_up(self) -> None:
178 if self.mcp_connection is not None:
179 return
180 self.mcp_connection = establish_connection(self.server_url)
181 self.tools = self.mcp_connection.fetch_tools()
182 ```
184 This method may be called multiple times (e.g. before every run): implementations are responsible for
185 their own idempotence, guarding on their own state as in the example above. The default implementation delegates
186 to the tools' own idempotent `warm_up()`.
187 """
188 for tool in self.tools:
189 if hasattr(tool, "warm_up"):
190 tool.warm_up()
192 def add(self, tool: "Tool | Toolset") -> None:
193 """
194 Add a new Tool or merge another Toolset.
196 Note: adding a Toolset flattens it into its individual tools, so this is only recommended
197 for Toolsets that don't manage shared resources in their `warm_up()` (or `__init__`).
198 For example, combining with an `MCPToolset`, which owns a shared connection, is not
199 recommended: the connection's lifecycle would no longer be managed by the original
200 Toolset.
202 Adding a Toolset is deprecated and will be removed in Haystack 3.2.0: pass Toolsets as a
203 list wherever tools are accepted instead, e.g. `Agent(tools=[toolset_a, toolset_b])`.
205 :param tool: A Tool instance or another Toolset to add
206 :raises ValueError: If adding the tool would result in duplicate tool names
207 :raises TypeError: If the provided object is not a Tool or Toolset
208 """
209 if not isinstance(tool, (Tool, Toolset)):
210 raise TypeError(f"Expected Tool or Toolset, got {type(tool).__name__}")
212 if isinstance(tool, Toolset):
213 warnings.warn(
214 "Adding a Toolset to another Toolset is deprecated and will be removed in Haystack 3.2.0. "
215 "Pass Toolsets as a list wherever tools are accepted instead, "
216 "e.g. Agent(tools=[toolset_a, toolset_b]).",
217 FutureWarning,
218 stacklevel=2,
219 )
221 new_tools = [tool] if isinstance(tool, Tool) else list(tool)
223 # Check for duplicates before adding
224 _check_duplicate_tool_names(self.tools + new_tools)
225 self.tools.extend(new_tools)
227 def to_dict(self) -> dict[str, Any]:
228 """
229 Serialize the Toolset to a dictionary.
231 :returns: A dictionary representation of the Toolset
233 Note for subclass implementers:
234 The default implementation is ideal for scenarios where Tool resolution is static. However, if your subclass
235 of Toolset dynamically resolves Tool instances from external sources—such as an MCP server, OpenAPI URL, or
236 a local OpenAPI specification—you should consider serializing the endpoint descriptor instead of the Tool
237 instances themselves. This strategy preserves the dynamic nature of your Toolset and minimizes the overhead
238 associated with serializing potentially large collections of Tool objects. Moreover, by serializing the
239 descriptor, you ensure that the deserialization process can accurately reconstruct the Tool instances, even
240 if they have been modified or removed since the last serialization. Failing to serialize the descriptor may
241 lead to issues where outdated or incorrect Tool configurations are loaded, potentially causing errors or
242 unexpected behavior.
243 """
244 return {
245 "type": generate_qualified_class_name(type(self)),
246 "data": {"tools": [tool.to_dict() for tool in self.tools]},
247 }
249 @classmethod
250 def from_dict(cls, data: dict[str, Any]) -> "Toolset":
251 """
252 Deserialize a Toolset from a dictionary.
254 :param data: Dictionary representation of the Toolset
255 :returns: A new Toolset instance
256 """
257 inner_data = data["data"]
258 tools_data = inner_data.get("tools", [])
260 tools = []
261 for tool_data in tools_data:
262 tool_class = import_class_by_name(tool_data["type"])
263 if not issubclass(tool_class, Tool):
264 raise TypeError(f"Class '{tool_class}' is not a subclass of Tool")
265 tools.append(tool_class.from_dict(tool_data))
267 return cls(tools=tools)
269 def __add__(self, other: "Tool | Toolset | list[Tool]") -> "Toolset":
270 """
271 Concatenate this Toolset with another Tool, Toolset, or list of Tools.
273 Deprecated: will be removed in Haystack 3.2.0. Pass tools and Toolsets as a list wherever tools
274 are accepted instead, e.g. `Agent(tools=[toolset_a, toolset_b])`.
276 :param other: Another Tool, Toolset, or list of Tools to concatenate
277 :returns: A new Toolset containing all tools
278 :raises TypeError: If the other parameter is not a Tool, Toolset, or list of Tools
279 :raises ValueError: If the combination would result in duplicate tool names
280 """
281 warnings.warn(
282 "Combining Toolsets and Tools with '+' is deprecated and will be removed in Haystack 3.2.0. "
283 "Pass them as a list wherever tools are accepted instead, e.g. Agent(tools=[toolset_a, toolset_b]).",
284 FutureWarning,
285 stacklevel=2,
286 )
287 if isinstance(other, Tool):
288 return Toolset(tools=self.tools + [other])
289 if isinstance(other, Toolset):
290 return _ToolsetWrapper([self, other])
291 if isinstance(other, list) and all(isinstance(item, Tool) for item in other):
292 return Toolset(tools=self.tools + other)
293 raise TypeError(f"Cannot add {type(other).__name__} to Toolset")
295 def __len__(self) -> int:
296 """
297 Return the number of Tools in this Toolset.
299 :returns: Number of Tools
300 """
301 return sum(1 for _ in self)
303 def __getitem__(self, index: int) -> Tool:
304 """
305 Get a Tool by index.
307 :param index: Index of the Tool to get
308 :returns: The Tool at the specified index
309 """
310 return list(self)[index]
313class _ToolsetWrapper(Toolset):
314 """
315 A wrapper that holds multiple toolsets and provides a unified interface.
317 This is used internally when combining different types of toolsets to preserve
318 their individual configurations while still being usable with Agent and Haystack chat generators.
320 Deprecated together with the `+` operator that creates it; both will be removed in Haystack 3.2.0.
321 """
323 def __init__(self, toolsets: list[Toolset]) -> None:
324 super().__init__([tool for toolset in toolsets for tool in toolset])
325 self.toolsets = toolsets
326 # Optional per-run name filter, set on the copies returned by spawn(). When set, iteration only
327 # yields tools whose name is in this set. None means no filtering.
328 self._selected_tool_names: set[str] | None = None
330 def __iter__(self) -> Iterator[Tool]:
331 """Iterate over all tools from all toolsets, honoring any active name filter."""
332 for toolset in self.toolsets:
333 for tool in toolset:
334 if self._selected_tool_names is None or tool.name in self._selected_tool_names:
335 yield tool
337 def get_selectable_tools(self) -> list[Tool]:
338 """Return every selectable tool across all wrapped toolsets, ignoring any active filter."""
339 return [tool for toolset in self.toolsets for tool in toolset.get_selectable_tools()]
341 def spawn(self, selected_tool_names: set[str] | None = None) -> "_ToolsetWrapper":
342 """
343 Return an isolated copy with each wrapped toolset spawned, carrying the given name selection.
345 :param selected_tool_names: Optional tool names this run is restricted to. None means no restriction.
346 :returns: A run-scoped copy of this wrapper.
347 """
348 new = _ToolsetWrapper([toolset.spawn(selected_tool_names=selected_tool_names) for toolset in self.toolsets])
349 new._selected_tool_names = set(selected_tool_names) if selected_tool_names is not None else None
350 return new
352 def __contains__(self, item: Any) -> bool:
353 """Check if a tool is in any of the toolsets."""
354 return any(item in toolset for toolset in self.toolsets)
356 def warm_up(self) -> None:
357 """Warm up all wrapped toolsets. May be called multiple times; the wrapped toolsets guard themselves."""
358 for toolset in self.toolsets:
359 toolset.warm_up()
361 def to_dict(self) -> dict[str, Any]:
362 """
363 Serialize the wrapper to a dictionary.
365 Each wrapped toolset is serialized via its own `to_dict()`, so any subclass that
366 overrides serialization (e.g. a toolset that serializes a connection/endpoint
367 descriptor) is preserved.
369 :returns: A dictionary representation of the wrapper.
370 """
371 return {
372 "type": generate_qualified_class_name(type(self)),
373 "data": {"toolsets": [toolset.to_dict() for toolset in self.toolsets]},
374 }
376 @classmethod
377 def from_dict(cls, data: dict[str, Any]) -> "_ToolsetWrapper":
378 """
379 Deserialize a wrapper from a dictionary.
381 :param data: Dictionary representation of the wrapper.
382 :returns: A new `_ToolsetWrapper` instance.
383 :raises TypeError: If any serialized entry is not a subclass of Toolset.
384 """
385 inner_data = data["data"]
386 toolsets_data = inner_data.get("toolsets", [])
388 toolsets = []
389 for toolset_data in toolsets_data:
390 toolset_class = import_class_by_name(toolset_data["type"])
391 if not issubclass(toolset_class, Toolset):
392 raise TypeError(f"Class '{toolset_class}' is not a subclass of Toolset")
393 toolsets.append(toolset_class.from_dict(toolset_data))
395 return cls(toolsets=toolsets)
397 def __add__(self, other: Toolset | Tool | list[Tool]) -> "_ToolsetWrapper":
398 """Add another toolset or tool to this wrapper. Deprecated, see `Toolset.__add__`."""
399 warnings.warn(
400 "Combining Toolsets and Tools with '+' is deprecated and will be removed in Haystack 3.2.0. "
401 "Pass them as a list wherever tools are accepted instead, e.g. Agent(tools=[toolset_a, toolset_b]).",
402 FutureWarning,
403 stacklevel=2,
404 )
405 if isinstance(other, Toolset):
406 return _ToolsetWrapper(self.toolsets + [other])
407 if isinstance(other, Tool):
408 return _ToolsetWrapper(self.toolsets + [Toolset([other])])
409 if isinstance(other, list) and all(isinstance(item, Tool) for item in other):
410 return _ToolsetWrapper(self.toolsets + [Toolset(other)])
411 raise TypeError(f"Cannot add {type(other).__name__} to _ToolsetWrapper")