Coverage for haystack/tools/toolset.py: 93%
46 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-09 14:34 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5from collections.abc import Iterator
6from dataclasses import dataclass, field
7from typing import Any
9from haystack.core.serialization import generate_qualified_class_name, import_class_by_name
10from haystack.tools.tool import Tool, _check_duplicate_tool_names
13@dataclass
14class Toolset:
15 """
16 A collection of related Tools that can be used and managed as a cohesive unit.
18 Toolset serves two main purposes:
20 1. Group related tools together:
21 Toolset allows you to organize related tools into a single collection, making it easier
22 to manage and use them as a unit in Haystack pipelines.
24 Example:
25 ```python
26 from typing import Annotated
27 from haystack.tools import tool, Toolset
28 from haystack.components.agents import Agent
29 from haystack.components.generators.chat import OpenAIResponsesChatGenerator
31 # Create tools with the @tool decorator (the recommended way)
32 @tool
33 def add(a: Annotated[int, "first number"], b: Annotated[int, "second number"]) -> int:
34 '''Add two numbers.'''
35 return a + b
37 @tool
38 def subtract(a: Annotated[int, "first number"], b: Annotated[int, "second number"]) -> int:
39 '''Subtract b from a.'''
40 return a - b
42 # Create a toolset with the math tools
43 math_toolset = Toolset([add, subtract])
45 # Use the toolset with an Agent
46 agent = Agent(chat_generator=OpenAIResponsesChatGenerator(), tools=math_toolset)
47 ```
49 2. Base class for dynamic tool loading:
50 By subclassing Toolset, you can create implementations that dynamically load tools from external sources like
51 OpenAPI URLs, MCP servers, or other resources.
53 When implementing a custom Toolset subclass for dynamic tool loading:
54 - Load the tools in `warm_up()` and assign them to `self.tools`. Since `warm_up()` may be called before
55 every run, make it idempotent by guarding on your own state (e.g. `if self._client is not None: return`).
56 - Override `to_dict()` and `from_dict()` to serialize the endpoint descriptor (URL, server info) rather than
57 the dynamically loaded Tool instances.
59 Example:
60 ```python
61 from haystack.core.serialization import generate_qualified_class_name
62 from haystack.tools import Toolset
64 class RemoteServiceToolset(Toolset):
65 def __init__(self, endpoint: str) -> None:
66 self.endpoint = endpoint
67 self._client = None
68 super().__init__(tools=[]) # tools are loaded on warm_up()
70 def warm_up(self) -> None:
71 if self._client is not None:
72 return
73 self._client = connect(self.endpoint)
74 self.tools = self._client.fetch_tools()
76 def to_dict(self):
77 return {
78 "type": generate_qualified_class_name(type(self)),
79 "data": {"endpoint": self.endpoint},
80 }
82 @classmethod
83 def from_dict(cls, data):
84 return cls(endpoint=data["data"]["endpoint"])
85 ```
87 Toolset implements the collection interface (__iter__, __contains__, __len__, __getitem__), making it behave like
88 a list of Tools. This makes it compatible with components that expect iterable tools, such as Agent or Haystack
89 chat generators.
90 """
92 # Use field() with default_factory to initialize the list
93 tools: list[Tool] = field(default_factory=list)
95 def __post_init__(self) -> None:
96 """
97 Validate the tools provided during initialization.
98 """
99 # If initialization was done a single Tool, raise an error
100 if isinstance(self.tools, Tool):
101 raise TypeError("A single Tool cannot be directly passed to Toolset. Please use a list: Toolset([tool])")
103 # Check for duplicate tool names in the initial set
104 _check_duplicate_tool_names(self.tools)
106 def __iter__(self) -> Iterator[Tool]:
107 """
108 Return an iterator over the Tools in this Toolset.
110 This allows the Toolset to be used wherever a list of Tools is expected.
112 :returns: An iterator yielding Tool instances
113 """
114 return iter(self.tools)
116 def get_selectable_tools(self) -> list[Tool]:
117 """
118 Return the tools available for name-based selection (e.g. via `Agent.run(tools=["tool_name"])`).
120 Prepare lazy toolsets with `warm_up_tools` or `warm_up_tools_async` before selection.
121 Subclasses whose iteration does not surface every selectable tool (e.g. SearchableToolset)
122 override this to return the full set.
124 :returns: The list of tools available for name-based selection.
125 """
126 return list(self.tools)
128 def spawn(self, selected_tool_names: set[str] | None = None) -> "Toolset": # noqa: ARG002
129 """
130 Return this Toolset, or an isolated copy of it, for a single run.
132 A plain Toolset has no run-scoped state, so the default implementation returns `self` and ignores the
133 selection (the Agent materializes it). Subclasses with run-scoped state (e.g. SearchableToolset) override
134 this to return a copy carrying the selection, so concurrent runs sharing the same configured Toolset
135 don't corrupt each other.
137 :param selected_tool_names: Optional tool names this run is restricted to. None means no restriction.
138 :returns: This Toolset, or a run-scoped copy of it.
139 """
140 return self
142 def __contains__(self, item: str | Tool) -> bool:
143 """
144 Check if a tool is in this Toolset.
146 Supports checking by:
147 - Tool instance: tool in toolset
148 - Tool name: "tool_name" in toolset
150 :param item: Tool instance or tool name string
151 :returns: True if contained, False otherwise
152 """
153 if isinstance(item, str):
154 return any(tool.name == item for tool in self)
155 if isinstance(item, Tool):
156 return any(tool is item or tool == item for tool in self)
157 return False
159 def add(self, tool: Tool) -> None:
160 """
161 Add a new Tool to this Toolset.
163 :param tool: A Tool instance to add
164 :raises ValueError: If adding the tool would result in duplicate tool names
165 :raises TypeError: If the provided object is not a Tool
166 """
167 if not isinstance(tool, Tool):
168 raise TypeError(f"Expected Tool, got {type(tool).__name__}")
170 # Check for duplicates before adding
171 _check_duplicate_tool_names(self.tools + [tool])
172 self.tools.append(tool)
174 def to_dict(self) -> dict[str, Any]:
175 """
176 Serialize the Toolset to a dictionary.
178 :returns: A dictionary representation of the Toolset
180 Note for subclass implementers:
181 The default implementation is ideal for scenarios where Tool resolution is static. However, if your subclass
182 of Toolset dynamically resolves Tool instances from external sources—such as an MCP server, OpenAPI URL, or
183 a local OpenAPI specification—you should consider serializing the endpoint descriptor instead of the Tool
184 instances themselves. This strategy preserves the dynamic nature of your Toolset and minimizes the overhead
185 associated with serializing potentially large collections of Tool objects. Moreover, by serializing the
186 descriptor, you ensure that the deserialization process can accurately reconstruct the Tool instances, even
187 if they have been modified or removed since the last serialization. Failing to serialize the descriptor may
188 lead to issues where outdated or incorrect Tool configurations are loaded, potentially causing errors or
189 unexpected behavior.
190 """
191 return {
192 "type": generate_qualified_class_name(type(self)),
193 "data": {"tools": [tool.to_dict() for tool in self.tools]},
194 }
196 @classmethod
197 def from_dict(cls, data: dict[str, Any]) -> "Toolset":
198 """
199 Deserialize a Toolset from a dictionary.
201 :param data: Dictionary representation of the Toolset
202 :returns: A new Toolset instance
203 """
204 inner_data = data["data"]
205 tools_data = inner_data.get("tools", [])
207 tools = []
208 for tool_data in tools_data:
209 tool_class = import_class_by_name(tool_data["type"])
210 if not issubclass(tool_class, Tool):
211 raise TypeError(f"Class '{tool_class}' is not a subclass of Tool")
212 tools.append(tool_class.from_dict(tool_data))
214 return cls(tools=tools)
216 def __len__(self) -> int:
217 """
218 Return the number of Tools in this Toolset.
220 :returns: Number of Tools
221 """
222 return sum(1 for _ in self)
224 def __getitem__(self, index: int) -> Tool:
225 """
226 Get a Tool by index.
228 :param index: Index of the Tool to get
229 :returns: The Tool at the specified index
230 """
231 return list(self)[index]