forked from strands-agents/sdk-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtools.py
More file actions
306 lines (221 loc) · 8.73 KB
/
tools.py
File metadata and controls
306 lines (221 loc) · 8.73 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
"""Tool-related type definitions for the SDK.
These types are modeled after the Bedrock API.
- Bedrock docs: https://docs.aws.amazon.com/bedrock/latest/APIReference/API_Types_Amazon_Bedrock_Runtime.html
"""
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, AsyncGenerator, Awaitable, Callable, Literal, Protocol, Union
from typing_extensions import NotRequired, TypedDict
from .interrupt import _Interruptible
from .media import DocumentContent, ImageContent
JSONSchema = dict
"""Type alias for JSON Schema dictionaries."""
class ToolSpec(TypedDict):
"""Specification for a tool that can be used by an agent.
Attributes:
description: A human-readable description of what the tool does.
inputSchema: JSON Schema defining the expected input parameters.
name: The unique name of the tool.
outputSchema: Optional JSON Schema defining the expected output format.
Note: Not all model providers support this field. Providers that don't
support it should filter it out before sending to their API.
"""
description: str
inputSchema: JSONSchema
name: str
outputSchema: NotRequired[JSONSchema]
class Tool(TypedDict):
"""A tool that can be provided to a model.
This type wraps a tool specification for inclusion in a model request.
Attributes:
toolSpec: The specification of the tool.
"""
toolSpec: ToolSpec
class ToolUse(TypedDict):
"""A request from the model to use a specific tool with the provided input.
Attributes:
input: The input parameters for the tool.
Can be any JSON-serializable type.
name: The name of the tool to invoke.
toolUseId: A unique identifier for this specific tool use request.
"""
input: Any
name: str
toolUseId: str
class ToolResultContent(TypedDict, total=False):
"""Content returned by a tool execution.
Attributes:
document: Document content returned by the tool.
image: Image content returned by the tool.
json: JSON-serializable data returned by the tool.
text: Text content returned by the tool.
"""
document: DocumentContent
image: ImageContent
json: Any
text: str
ToolResultStatus = Literal["success", "error"]
"""Status of a tool execution result."""
class ToolResult(TypedDict):
"""Result of a tool execution.
Attributes:
content: List of result content returned by the tool.
status: The status of the tool execution ("success" or "error").
toolUseId: The unique identifier of the tool use request that produced this result.
"""
content: list[ToolResultContent]
status: ToolResultStatus
toolUseId: str
class ToolChoiceAuto(TypedDict):
"""Configuration for automatic tool selection.
This represents the configuration for automatic tool selection, where the model decides whether and which tool to
use based on the context.
"""
pass
class ToolChoiceAny(TypedDict):
"""Configuration indicating that the model must request at least one tool."""
pass
class ToolChoiceTool(TypedDict):
"""Configuration for forcing the use of a specific tool.
Attributes:
name: The name of the tool that the model must use.
"""
name: str
@dataclass
class ToolContext(_Interruptible):
"""Context object containing framework-provided data for decorated tools.
This object provides access to framework-level information that may be useful
for tool implementations.
Attributes:
tool_use: The complete ToolUse object containing tool invocation details.
agent: The Agent or BidiAgent instance executing this tool, providing access to conversation history,
model configuration, and other agent state.
invocation_state: Caller-provided kwargs that were passed to the agent when it was invoked (agent(),
agent.invoke_async(), etc.).
Note:
This class is intended to be instantiated by the SDK. Direct construction by users
is not supported and may break in future versions as new fields are added.
"""
tool_use: ToolUse
agent: Any # Agent or BidiAgent - using Any for backwards compatibility
invocation_state: dict[str, Any]
def _interrupt_id(self, name: str) -> str:
"""Unique id for the interrupt.
Args:
name: User defined name for the interrupt.
Returns:
Interrupt id.
"""
return f"v1:tool_call:{self.tool_use['toolUseId']}:{uuid.uuid5(uuid.NAMESPACE_OID, name)}"
# Individual ToolChoice type aliases
ToolChoiceAutoDict = dict[Literal["auto"], ToolChoiceAuto]
ToolChoiceAnyDict = dict[Literal["any"], ToolChoiceAny]
ToolChoiceToolDict = dict[Literal["tool"], ToolChoiceTool]
ToolChoice = Union[
ToolChoiceAutoDict,
ToolChoiceAnyDict,
ToolChoiceToolDict,
]
"""
Configuration for how the model should choose tools.
- "auto": The model decides whether to use tools based on the context
- "any": The model must use at least one tool (any tool)
- "tool": The model must use the specified tool
"""
RunToolHandler = Callable[[ToolUse], AsyncGenerator[dict[str, Any], None]]
"""Callback that runs a single tool and streams back results."""
ToolGenerator = AsyncGenerator[Any, None]
"""Generator of tool events with the last being the tool result."""
class ToolConfig(TypedDict):
"""Configuration for tools in a model request.
Attributes:
tools: List of tools available to the model.
toolChoice: Configuration for how the model should choose tools.
"""
tools: list[Tool]
toolChoice: ToolChoice
class ToolFunc(Protocol):
"""Function signature for Python decorated and module based tools."""
__name__: str
def __call__(
self, *args: Any, **kwargs: Any
) -> Union[
ToolResult,
Awaitable[ToolResult],
]:
"""Function signature for Python decorated and module based tools.
Returns:
Tool result or awaitable tool result.
"""
...
class AgentTool(ABC):
"""Abstract base class for all SDK tools.
This class defines the interface that all tool implementations must follow. Each tool must provide its name,
specification, and implement a stream method that executes the tool's functionality.
"""
_is_dynamic: bool
def __init__(self) -> None:
"""Initialize the base agent tool with default dynamic state."""
self._is_dynamic = False
@property
@abstractmethod
# pragma: no cover
def tool_name(self) -> str:
"""The unique name of the tool used for identification and invocation."""
pass
@property
@abstractmethod
# pragma: no cover
def tool_spec(self) -> ToolSpec:
"""Tool specification that describes its functionality and parameters."""
pass
@property
@abstractmethod
# pragma: no cover
def tool_type(self) -> str:
"""The type of the tool implementation (e.g., 'python', 'javascript', 'lambda').
Used for categorization and appropriate handling.
"""
pass
@property
def supports_hot_reload(self) -> bool:
"""Whether the tool supports automatic reloading when modified.
Returns:
False by default.
"""
return False
@abstractmethod
# pragma: no cover
def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
"""Stream tool events and return the final result.
Args:
tool_use: The tool use request containing tool ID and parameters.
invocation_state: Caller-provided kwargs that were passed to the agent when it was invoked (agent(),
agent.invoke_async(), etc.).
**kwargs: Additional keyword arguments for future extensibility.
Yields:
Tool events with the last being the tool result.
"""
...
@property
def is_dynamic(self) -> bool:
"""Whether the tool was dynamically loaded during runtime.
Dynamic tools may have different lifecycle management.
Returns:
True if loaded dynamically, False otherwise.
"""
return self._is_dynamic
def mark_dynamic(self) -> None:
"""Mark this tool as dynamically loaded."""
self._is_dynamic = True
def get_display_properties(self) -> dict[str, str]:
"""Get properties to display in UI representations of this tool.
Subclasses can extend this to include additional properties.
Returns:
Dictionary of property names and their string values.
"""
return {
"Name": self.tool_name,
"Type": self.tool_type,
}