Coverage for haystack/tools/agent_tool.py: 100%
51 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
5import json
6from collections.abc import Callable
7from typing import Any
9from haystack.components.agents import Agent
10from haystack.components.agents.agent import _EXIT_REASON_CONTENT_FILTER, _EXIT_REASON_LENGTH, _EXIT_REASON_MAX_STEPS
11from haystack.tools.component_tool import ComponentTool
12from haystack.tools.tool import _deserialize_outputs_to_state, _deserialize_outputs_to_string
13from haystack.utils.deserialization import deserialize_component_inplace
16def _required_tool_parameters(agent: Agent, inputs_from_state: dict[str, Any] | None) -> list[str]:
17 """
18 Return the additional required Tool parameters for the wrapped Agent.
20 These are mandatory Agent inputs that are not supplied through `inputs_from_state`.
21 `messages` is excluded because AgentTool always adds it to the generated schema.
23 :param agent: The wrapped Agent.
24 :param inputs_from_state: Maps the calling Agent's state keys to Agent inputs.
25 :returns: Additional required Tool parameters, sorted by name.
26 """
27 covered = {"messages", *(inputs_from_state or {}).values()}
28 # prompt variables are registered as input sockets in a non-deterministic order, so sort to keep the schema stable
29 return sorted(
30 name
31 # __haystack_input__ is attached to the instance by the @component decorator, so mypy cannot see it
32 for name, socket in agent.__haystack_input__._sockets_dict.items() # type: ignore[attr-defined]
33 if socket.is_mandatory and name not in covered
34 )
37def agent_result_to_string(result: dict[str, Any]) -> str:
38 """Default `outputs_to_string` handler"""
39 last_message = result["last_message"]
40 text = last_message.text or json.dumps(last_message.to_dict())
41 exit_reason = result["exit_reason"]
42 if exit_reason == _EXIT_REASON_MAX_STEPS:
43 text += "\n\n[The Agent reached max_agent_steps and stopped, so this result may be incomplete.]"
44 elif exit_reason == _EXIT_REASON_LENGTH:
45 text += "\n\n[The Agent reached the model's output limit and stopped, so this result may be incomplete.]"
46 elif exit_reason == _EXIT_REASON_CONTENT_FILTER:
47 text += "\n\n[The Agent's response was stopped by a content filter, so this result may be incomplete.]"
48 return text
51class AgentTool(ComponentTool):
52 """
53 A Tool that wraps a Haystack Agent, allowing it to be used as a tool by another Agent.
55 AgentTool is a building block for multi-agent systems: an Agent specialized in one task becomes a tool that
56 other Agents can delegate to. The calling Agent only sees the final reply, so all the steps the wrapped Agent
57 takes stay out of its context. Sensible defaults make this work out of the box: the task is delegated as a
58 single user message and comes back as text.
60 To use AgentTool, you first need a Haystack Agent. Below is an example of creating an AgentTool from an Agent
61 that searches the web with a SerperDevWebSearch component from the `serperdev-haystack` integration package
62 (`pip install serperdev-haystack`).
64 ## Usage Example:
65 <!-- test-ignore -->
66 ```python
67 from haystack.components.agents import Agent
68 from haystack.components.generators.chat import OpenAIResponsesChatGenerator
69 from haystack.dataclasses import ChatMessage
70 from haystack.tools import AgentTool, ComponentTool
71 from haystack_integrations.components.websearch.serperdev import SerperDevWebSearch
73 researcher = Agent(
74 chat_generator=OpenAIResponsesChatGenerator(model="gpt-5.4-mini"),
75 system_prompt="You are a research specialist. Investigate the task and report your findings.",
76 tools=[
77 ComponentTool(
78 component=SerperDevWebSearch(
79 top_k=3,
80 ),
81 name="web_search",
82 description="Search the web for current information on any topic",
83 ),
84 ],
85 )
87 research = AgentTool(
88 agent=researcher,
89 name="research",
90 description="Research a question on the web and report the findings",
91 )
93 coordinator = Agent(
94 chat_generator=OpenAIResponsesChatGenerator(model="gpt-5.4"),
95 tools=[research],
96 system_prompt="You coordinate specialists. Delegate research questions, then answer the user.",
97 )
99 result = coordinator.run([ChatMessage.from_user("What are the latest developments in the Haystack framework?")])
100 print(result["last_message"].text)
101 ```
102 """
104 def __init__(
105 self,
106 agent: Agent,
107 *,
108 name: str,
109 description: str,
110 parameters: dict[str, Any] | None = None,
111 outputs_to_string: dict[str, str | Callable[[Any], str]] | None = None,
112 inputs_from_state: dict[str, str] | None = None,
113 outputs_to_state: dict[str, dict[str, str | Callable]] | None = None,
114 ) -> None:
115 """
116 Create a Tool instance from a Haystack Agent.
118 :param agent: The Haystack Agent to wrap as a tool.
119 :param name: Name of the tool.
120 :param description: Description of the tool. It should tell the calling LLM what the Agent is specialized in
121 and when to delegate to it.
122 :param parameters:
123 A JSON schema defining the parameters expected by the Tool.
124 Will fall back to a schema with the task to delegate as a single user message, plus one string parameter
125 for every other mandatory input of the Agent, if not provided.
126 :param outputs_to_string:
127 Optional dictionary defining how tool outputs should be converted into string(s) or results.
128 If not provided, the tool result is the text of the Agent's final reply, or the serialized message if
129 the reply has no text. A warning is appended if the Agent stopped because it reached `max_agent_steps`,
130 reached the model's output limit, or had its response stopped by a content filter.
132 `outputs_to_string` supports two formats:
134 1. Single output format - use "source", "handler", and/or "raw_result" at the root level:
135 ```python
136 {
137 "source": "last_message", "handler": format_reply, "raw_result": False
138 }
139 ```
140 - `source`: If provided, only the specified output key is sent to the handler.
141 - `handler`: A function that takes the tool output (or the extracted source value) and returns the
142 final result.
143 - `raw_result`: If `True`, the result is returned raw without string conversion, but applying the
144 `handler` if provided. This is intended for tools that return images. In this mode, the `handler`
145 is required, since the Agent returns a dictionary, and it must return a list of
146 `TextContent`/`ImageContent` objects to ensure compatibility with Chat Generators.
148 2. Multiple output format - map keys to individual configurations:
149 ```python
150 {
151 "reply": {"source": "last_message", "handler": format_reply},
152 "steps": {"source": "step_count", "handler": str}
153 }
154 ```
155 Each key maps to a dictionary that can contain "source" and/or "handler".
156 Note that `raw_result` is not supported in the multiple output format.
157 :param inputs_from_state:
158 Optional dictionary mapping the calling Agent's state keys to Agent input names.
159 Example: `{"subject": "topic"}` maps state's "subject" to the Agent's "topic" input.
160 Inputs mapped this way are not added to the generated `parameters` schema, since the calling Agent
161 provides them.
162 :param outputs_to_state:
163 Optional dictionary defining how tool outputs map to keys within state as well as optional handlers.
164 The keys must be declared in the `state_schema` of the calling Agent.
165 Handlers merge the tool output into the state and are called as `handler(current_value, tool_output)`.
166 If the source is provided only the specified output key is sent to the handler.
167 Example:
168 ```python
169 {
170 "notes": {"source": "last_message", "handler": custom_handler}
171 }
172 ```
173 If the source is omitted the whole tool result is sent to the handler.
174 Example:
175 ```python
176 {
177 "notes": {"handler": custom_handler}
178 }
179 ```
180 :raises TypeError: If the object passed is not a Haystack Agent instance.
181 :raises ValueError: If `parameters` is provided but does not cover all the mandatory inputs of the Agent.
182 """
183 if not isinstance(agent, Agent):
184 raise TypeError(f"The 'agent' parameter must be an instance of Agent. Got {type(agent)} instead.")
186 if parameters is not None:
187 required_tool_parameters = _required_tool_parameters(agent=agent, inputs_from_state=inputs_from_state)
188 missing_required_parameters = [
189 name for name in required_tool_parameters if name not in parameters.get("properties", {})
190 ]
191 if missing_required_parameters:
192 raise ValueError(
193 f"The Agent wrapped by this tool requires the inputs {missing_required_parameters}, but this tool "
194 f"does not supply them, so it can never run. Add them to 'parameters', the schema that the calling "
195 f"LLM fills in, or to 'inputs_from_state', which takes them from the calling Agent's state."
196 )
198 super().__init__(
199 component=agent,
200 name=name,
201 description=description,
202 parameters=parameters,
203 outputs_to_string=outputs_to_string or {"handler": agent_result_to_string},
204 inputs_from_state=inputs_from_state,
205 outputs_to_state=outputs_to_state,
206 )
208 def _create_tool_parameters_schema(self, component: Any, inputs_from_state: dict[str, Any]) -> dict[str, Any]:
209 """
210 Override ComponentTool schema generation for AgentTool defaults.
212 ComponentTool calls this when users do not provide an explicit `parameters` schema. The generated schema always
213 includes `messages` for the delegated task, plus one string parameter for each mandatory Agent input not
214 supplied through `inputs_from_state`.
215 """
216 additional_required_parameters = _required_tool_parameters(agent=component, inputs_from_state=inputs_from_state)
217 additional_properties = {name: {"type": "string"} for name in additional_required_parameters}
218 return {
219 "type": "object",
220 "properties": {
221 "messages": {
222 "type": "array",
223 "description": "Exactly one user message.",
224 "minItems": 1,
225 "maxItems": 1,
226 "items": {
227 "type": "object",
228 "properties": {
229 "role": {"type": "string", "enum": ["user"]},
230 "content": {"type": "string", "description": "The task to delegate to this tool."},
231 },
232 "required": ["role", "content"],
233 },
234 },
235 **additional_properties,
236 },
237 "required": ["messages", *additional_required_parameters],
238 }
240 def to_dict(self) -> dict[str, Any]:
241 """
242 Serializes the AgentTool to a dictionary.
244 :returns:
245 The serialized dictionary representation of AgentTool.
246 """
247 serialized = super().to_dict()
248 serialized["data"]["agent"] = serialized["data"].pop("component")
249 return serialized
251 @classmethod
252 def from_dict(cls, data: dict[str, Any]) -> "AgentTool":
253 """
254 Deserializes the AgentTool from a dictionary.
256 :param data: The dictionary representation of AgentTool.
257 :returns:
258 The deserialized AgentTool instance.
259 """
260 inner_data = data["data"]
261 deserialize_component_inplace(data=inner_data, key="agent")
263 outputs_to_state = inner_data.get("outputs_to_state")
264 if outputs_to_state:
265 outputs_to_state = _deserialize_outputs_to_state(outputs_to_state=outputs_to_state)
267 outputs_to_string = inner_data.get("outputs_to_string")
268 if outputs_to_string is not None:
269 outputs_to_string = _deserialize_outputs_to_string(outputs_to_string=outputs_to_string)
271 return cls(
272 agent=inner_data["agent"],
273 name=inner_data["name"],
274 description=inner_data["description"],
275 parameters=inner_data.get("parameters"),
276 outputs_to_string=outputs_to_string,
277 inputs_from_state=inner_data.get("inputs_from_state"),
278 outputs_to_state=outputs_to_state,
279 )