18532846500

Committed 15 Oct 2025 02:45PM UTC coverage: 92.115% (+0.01%) from 92.103%

Build # 18532846500

Build Type

Pull #9880

github

Committed by

web-flow

Commit Message

Merge aacbda1c8 into 91814326f

Pull Request Pull Request #9880: draft: Expand tools param to include list[Toolset]

Run Details

13283 of 14420 relevant lines covered (92.12%)

0.92 hits per line

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

96.47

haystack/tools/tool.py

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

from dataclasses import asdict, dataclass
from typing import Any, Callable, Optional

from jsonschema import Draft202012Validator
from jsonschema.exceptions import SchemaError

from haystack.core.serialization import generate_qualified_class_name
from haystack.tools.errors import ToolInvocationError
from haystack.utils.callable_serialization import deserialize_callable, serialize_callable


@dataclass
class Tool:
    """
    Data class representing a Tool that Language Models can prepare a call for.

    Accurate definitions of the textual attributes such as `name` and `description`
    are important for the Language Model to correctly prepare the call.

    :param name:
        Name of the Tool.
    :param description:
        Description of the Tool.
    :param parameters:
        A JSON schema defining the parameters expected by the Tool.
    :param function:
        The function that will be invoked when the Tool is called.
    :param outputs_to_string:
        Optional dictionary defining how a tool outputs should be converted into a string.
        If the source is provided only the specified output key is sent to the handler.
        If the source is omitted the whole tool result is sent to the handler.
        Example:
        ```python
        {
            "source": "docs", "handler": format_documents
        }
        ```
    :param inputs_from_state:
        Optional dictionary mapping state keys to tool parameter names.
        Example: `{"repository": "repo"}` maps state's "repository" to tool's "repo" parameter.
    :param outputs_to_state:
        Optional dictionary defining how tool outputs map to keys within state as well as optional handlers.
        If the source is provided only the specified output key is sent to the handler.
        Example:
        ```python
        {
            "documents": {"source": "docs", "handler": custom_handler}
        }
        ```
        If the source is omitted the whole tool result is sent to the handler.
        Example:
        ```python
        {
            "documents": {"handler": custom_handler}
        }
        ```
    """

    name: str
    description: str
    parameters: dict[str, Any]
    function: Callable
    outputs_to_string: Optional[dict[str, Any]] = None
    inputs_from_state: Optional[dict[str, str]] = None
    outputs_to_state: Optional[dict[str, dict[str, Any]]] = None

    def __post_init__(self):
        # Check that the parameters define a valid JSON schema
        try:
            Draft202012Validator.check_schema(self.parameters)
        except SchemaError as e:
            raise ValueError("The provided parameters do not define a valid JSON schema") from e

        # Validate outputs structure if provided
        if self.outputs_to_state is not None:
            for key, config in self.outputs_to_state.items():
                if not isinstance(config, dict):
                    raise ValueError(f"outputs_to_state configuration for key '{key}' must be a dictionary")
                if "source" in config and not isinstance(config["source"], str):
                    raise ValueError(f"outputs_to_state source for key '{key}' must be a string.")
                if "handler" in config and not callable(config["handler"]):
                    raise ValueError(f"outputs_to_state handler for key '{key}' must be callable")

        if self.outputs_to_string is not None:
            if "source" in self.outputs_to_string and not isinstance(self.outputs_to_string["source"], str):
                raise ValueError("outputs_to_string source must be a string.")
            if "handler" in self.outputs_to_string and not callable(self.outputs_to_string["handler"]):
                raise ValueError("outputs_to_string handler must be callable")

    @property
    def tool_spec(self) -> dict[str, Any]:
        """
        Return the Tool specification to be used by the Language Model.
        """
        return {"name": self.name, "description": self.description, "parameters": self.parameters}

    def invoke(self, **kwargs: Any) -> Any:
        """
        Invoke the Tool with the provided keyword arguments.
        """
        try:
            result = self.function(**kwargs)
        except Exception as e:
            raise ToolInvocationError(
                f"Failed to invoke Tool `{self.name}` with parameters {kwargs}. Error: {e}", tool_name=self.name
            ) from e
        return result

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

        :returns:
            Dictionary with serialized data.
        """
        data = asdict(self)
        data["function"] = serialize_callable(self.function)

        if self.outputs_to_state is not None:
            data["outputs_to_state"] = _serialize_outputs_to_state(self.outputs_to_state)

        if self.outputs_to_string is not None and self.outputs_to_string.get("handler") is not None:
            # This is soft-copied as to not modify the attributes in place
            data["outputs_to_string"] = self.outputs_to_string.copy()
            data["outputs_to_string"]["handler"] = serialize_callable(self.outputs_to_string["handler"])
        else:
            data["outputs_to_string"] = None

        return {"type": generate_qualified_class_name(type(self)), "data": data}

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

        :param data:
            Dictionary to deserialize from.
        :returns:
            Deserialized Tool.
        """
        init_parameters = data["data"]
        init_parameters["function"] = deserialize_callable(init_parameters["function"])
        if "outputs_to_state" in init_parameters and init_parameters["outputs_to_state"]:
            init_parameters["outputs_to_state"] = _deserialize_outputs_to_state(init_parameters["outputs_to_state"])

        if (
            init_parameters.get("outputs_to_string") is not None
            and init_parameters["outputs_to_string"].get("handler") is not None
        ):
            init_parameters["outputs_to_string"]["handler"] = deserialize_callable(
                init_parameters["outputs_to_string"]["handler"]
            )

        return cls(**init_parameters)


def _check_duplicate_tool_names(tools: Optional[list[Tool]]) -> None:
    """
    Checks for duplicate tool names and raises a ValueError if they are found.

    :param tools: The list of tools to check.
    :raises ValueError: If duplicate tool names are found.
    """
    if tools is None:
        return
    tool_names = [tool.name for tool in tools]
    duplicate_tool_names = {name for name in tool_names if tool_names.count(name) > 1}
    if duplicate_tool_names:
        raise ValueError(f"Duplicate tool names found: {duplicate_tool_names}")


def _serialize_outputs_to_state(outputs_to_state: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]:
    """
    Serializes the outputs_to_state dictionary, converting any callable handlers to their string representation.

    :param outputs_to_state: The outputs_to_state dictionary to serialize.
    :returns: The serialized outputs_to_state dictionary.
    """
    serialized_outputs = {}
    for key, config in outputs_to_state.items():
        serialized_config = config.copy()
        if "handler" in config:
            serialized_config["handler"] = serialize_callable(config["handler"])
        serialized_outputs[key] = serialized_config
    return serialized_outputs


def _deserialize_outputs_to_state(outputs_to_state: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]:
    """
    Deserializes the outputs_to_state dictionary, converting any string handlers back to callables.

    :param outputs_to_state: The outputs_to_state dictionary to deserialize.
    :returns: The deserialized outputs_to_state dictionary.
    """
    deserialized_outputs = {}
    for key, config in outputs_to_state.items():
        deserialized_config = config.copy()
        if "handler" in config:
            deserialized_config["handler"] = deserialize_callable(config["handler"])
        deserialized_outputs[key] = deserialized_config
    return deserialized_outputs

1	# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2	#
3	# SPDX-License-Identifier: Apache-2.0
4
5	from dataclasses import asdict, dataclass	1✔
6	from typing import Any, Callable, Optional	1✔
7
8	from jsonschema import Draft202012Validator	1✔
9	from jsonschema.exceptions import SchemaError	1✔
10
11	from haystack.core.serialization import generate_qualified_class_name	1✔
12	from haystack.tools.errors import ToolInvocationError	1✔
13	from haystack.utils.callable_serialization import deserialize_callable, serialize_callable	1✔
14
15
16	@dataclass	1✔
17	class Tool:	1✔
18	"""
19	Data class representing a Tool that Language Models can prepare a call for.
20
21	Accurate definitions of the textual attributes such as `name` and `description`
22	are important for the Language Model to correctly prepare the call.
23
24	:param name:
25	Name of the Tool.
26	:param description:
27	Description of the Tool.
28	:param parameters:
29	A JSON schema defining the parameters expected by the Tool.
30	:param function:
31	The function that will be invoked when the Tool is called.
32	:param outputs_to_string:
33	Optional dictionary defining how a tool outputs should be converted into a string.
34	If the source is provided only the specified output key is sent to the handler.
35	If the source is omitted the whole tool result is sent to the handler.
36	Example:
37	```python
38	{
39	"source": "docs", "handler": format_documents
40	}
41	```
42	:param inputs_from_state:
43	Optional dictionary mapping state keys to tool parameter names.
44	Example: `{"repository": "repo"}` maps state's "repository" to tool's "repo" parameter.
45	:param outputs_to_state:
46	Optional dictionary defining how tool outputs map to keys within state as well as optional handlers.
47	If the source is provided only the specified output key is sent to the handler.
48	Example:
49	```python
50	{
51	"documents": {"source": "docs", "handler": custom_handler}
52	}
53	```
54	If the source is omitted the whole tool result is sent to the handler.
55	Example:
56	```python
57	{
58	"documents": {"handler": custom_handler}
59	}
60	```
61	"""
62
63	name: str	1✔
64	description: str	1✔
65	parameters: dict[str, Any]	1✔
66	function: Callable	1✔
67	outputs_to_string: Optional[dict[str, Any]] = None	1✔
68	inputs_from_state: Optional[dict[str, str]] = None	1✔
69	outputs_to_state: Optional[dict[str, dict[str, Any]]] = None	1✔
70
71	def __post_init__(self):	1✔
72	# Check that the parameters define a valid JSON schema
73	try:	1✔
74	Draft202012Validator.check_schema(self.parameters)	1✔
75	except SchemaError as e:	1✔
76	raise ValueError("The provided parameters do not define a valid JSON schema") from e	1✔
77
78	# Validate outputs structure if provided
79	if self.outputs_to_state is not None:	1✔
80	for key, config in self.outputs_to_state.items():	1✔
81	if not isinstance(config, dict):	1✔
82	raise ValueError(f"outputs_to_state configuration for key '{key}' must be a dictionary")	1✔
83	if "source" in config and not isinstance(config["source"], str):	1✔
84	raise ValueError(f"outputs_to_state source for key '{key}' must be a string.")	1✔
85	if "handler" in config and not callable(config["handler"]):	1✔
86	raise ValueError(f"outputs_to_state handler for key '{key}' must be callable")	1✔
87
88	if self.outputs_to_string is not None:	1✔
89	if "source" in self.outputs_to_string and not isinstance(self.outputs_to_string["source"], str):	1✔
90	raise ValueError("outputs_to_string source must be a string.")	×
91	if "handler" in self.outputs_to_string and not callable(self.outputs_to_string["handler"]):	1✔
92	raise ValueError("outputs_to_string handler must be callable")	×
93
94	@property	1✔
95	def tool_spec(self) -> dict[str, Any]:	1✔
96	"""
97	Return the Tool specification to be used by the Language Model.
98	"""
99	return {"name": self.name, "description": self.description, "parameters": self.parameters}	1✔
100
101	def invoke(self, **kwargs: Any) -> Any:	1✔
102	"""
103	Invoke the Tool with the provided keyword arguments.
104	"""
105	try:	1✔
106	result = self.function(**kwargs)	1✔
107	except Exception as e:	1✔
108	raise ToolInvocationError(	1✔
109	f"Failed to invoke Tool `{self.name}` with parameters {kwargs}. Error: {e}", tool_name=self.name
110	) from e
111	return result	1✔
112
113	def to_dict(self) -> dict[str, Any]:	1✔
114	"""
115	Serializes the Tool to a dictionary.
116
117	:returns:
118	Dictionary with serialized data.
119	"""
120	data = asdict(self)	1✔
121	data["function"] = serialize_callable(self.function)	1✔
122
123	if self.outputs_to_state is not None:	1✔
124	data["outputs_to_state"] = _serialize_outputs_to_state(self.outputs_to_state)	1✔
125
126	if self.outputs_to_string is not None and self.outputs_to_string.get("handler") is not None:	1✔
127	# This is soft-copied as to not modify the attributes in place
128	data["outputs_to_string"] = self.outputs_to_string.copy()	1✔
129	data["outputs_to_string"]["handler"] = serialize_callable(self.outputs_to_string["handler"])	1✔
130	else:
131	data["outputs_to_string"] = None	1✔
132
133	return {"type": generate_qualified_class_name(type(self)), "data": data}	1✔
134
135	@classmethod	1✔
136	def from_dict(cls, data: dict[str, Any]) -> "Tool":	1✔
137	"""
138	Deserializes the Tool from a dictionary.
139
140	:param data:
141	Dictionary to deserialize from.
142	:returns:
143	Deserialized Tool.
144	"""
145	init_parameters = data["data"]	1✔
146	init_parameters["function"] = deserialize_callable(init_parameters["function"])	1✔
147	if "outputs_to_state" in init_parameters and init_parameters["outputs_to_state"]:	1✔
148	init_parameters["outputs_to_state"] = _deserialize_outputs_to_state(init_parameters["outputs_to_state"])	1✔
149
150	if (	1✔
151	init_parameters.get("outputs_to_string") is not None
152	and init_parameters["outputs_to_string"].get("handler") is not None
153	):
154	init_parameters["outputs_to_string"]["handler"] = deserialize_callable(	1✔
155	init_parameters["outputs_to_string"]["handler"]
156	)
157
158	return cls(**init_parameters)	1✔
159
160
161	def _check_duplicate_tool_names(tools: Optional[list[Tool]]) -> None:	1✔
162	"""
163	Checks for duplicate tool names and raises a ValueError if they are found.
164
165	:param tools: The list of tools to check.
166	:raises ValueError: If duplicate tool names are found.
167	"""
168	if tools is None:	1✔
169	return	×
170	tool_names = [tool.name for tool in tools]	1✔
171	duplicate_tool_names = {name for name in tool_names if tool_names.count(name) > 1}	1✔
172	if duplicate_tool_names:	1✔
173	raise ValueError(f"Duplicate tool names found: {duplicate_tool_names}")	1✔
174
175
176	def _serialize_outputs_to_state(outputs_to_state: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]:	1✔
177	"""
178	Serializes the outputs_to_state dictionary, converting any callable handlers to their string representation.
179
180	:param outputs_to_state: The outputs_to_state dictionary to serialize.
181	:returns: The serialized outputs_to_state dictionary.
182	"""
183	serialized_outputs = {}	1✔
184	for key, config in outputs_to_state.items():	1✔
185	serialized_config = config.copy()	1✔
186	if "handler" in config:	1✔
187	serialized_config["handler"] = serialize_callable(config["handler"])	1✔
188	serialized_outputs[key] = serialized_config	1✔
189	return serialized_outputs	1✔
190
191
192	def _deserialize_outputs_to_state(outputs_to_state: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]:	1✔
193	"""
194	Deserializes the outputs_to_state dictionary, converting any string handlers back to callables.
195
196	:param outputs_to_state: The outputs_to_state dictionary to deserialize.
197	:returns: The deserialized outputs_to_state dictionary.
198	"""
199	deserialized_outputs = {}	1✔
200	for key, config in outputs_to_state.items():	1✔
201	deserialized_config = config.copy()	1✔
202	if "handler" in config:	1✔
203	deserialized_config["handler"] = deserialize_callable(config["handler"])	1✔
204	deserialized_outputs[key] = deserialized_config	1✔
205	return deserialized_outputs	1✔

deepset-ai / haystack / 18532846500

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