Skip to content

strands.tools.structured_output.structured_output_tool

Structured output tool implementation.

This module provides a real tool implementation for structured output that integrates with the existing tool execution and error handling infrastructure.

ToolGenerator = AsyncGenerator[Any, None] module-attribute

Generator of tool events with the last being the tool result.

_TOOL_SPEC_CACHE = {} module-attribute

logger = logging.getLogger(__name__) module-attribute

AgentTool

Bases: 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.

Source code in strands/types/tools.py
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
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,
        }

is_dynamic property

Whether the tool was dynamically loaded during runtime.

Dynamic tools may have different lifecycle management.

Returns:

Type Description
bool

True if loaded dynamically, False otherwise.

supports_hot_reload property

Whether the tool supports automatic reloading when modified.

Returns:

Type Description
bool

False by default.

tool_name abstractmethod property

The unique name of the tool used for identification and invocation.

tool_spec abstractmethod property

Tool specification that describes its functionality and parameters.

tool_type abstractmethod property

The type of the tool implementation (e.g., 'python', 'javascript', 'lambda').

Used for categorization and appropriate handling.

__init__()

Initialize the base agent tool with default dynamic state.

Source code in strands/types/tools.py
227
228
229
def __init__(self) -> None:
    """Initialize the base agent tool with default dynamic state."""
    self._is_dynamic = False

get_display_properties()

Get properties to display in UI representations of this tool.

Subclasses can extend this to include additional properties.

Returns:

Type Description
dict[str, str]

Dictionary of property names and their string values.

Source code in strands/types/tools.py
295
296
297
298
299
300
301
302
303
304
305
306
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,
    }

mark_dynamic()

Mark this tool as dynamically loaded.

Source code in strands/types/tools.py
291
292
293
def mark_dynamic(self) -> None:
    """Mark this tool as dynamically loaded."""
    self._is_dynamic = True

stream(tool_use, invocation_state, **kwargs) abstractmethod

Stream tool events and return the final result.

Parameters:

Name Type Description Default
tool_use ToolUse

The tool use request containing tool ID and parameters.

required
invocation_state dict[str, Any]

Caller-provided kwargs that were passed to the agent when it was invoked (agent(), agent.invoke_async(), etc.).

required
**kwargs Any

Additional keyword arguments for future extensibility.

{}

Yields:

Type Description
ToolGenerator

Tool events with the last being the tool result.

Source code in strands/types/tools.py
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
@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.
    """
    ...

StructuredOutputContext

Per-invocation context for structured output execution.

Source code in strands/tools/structured_output/_structured_output_context.py
 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
class StructuredOutputContext:
    """Per-invocation context for structured output execution."""

    def __init__(self, structured_output_model: Type[BaseModel] | None = None):
        """Initialize a new structured output context.

        Args:
            structured_output_model: Optional Pydantic model type for structured output.
        """
        self.results: dict[str, BaseModel] = {}
        self.structured_output_model: Type[BaseModel] | None = structured_output_model
        self.structured_output_tool: StructuredOutputTool | None = None
        self.forced_mode: bool = False
        self.force_attempted: bool = False
        self.tool_choice: ToolChoice | None = None
        self.stop_loop: bool = False
        self.expected_tool_name: Optional[str] = None

        if structured_output_model:
            self.structured_output_tool = StructuredOutputTool(structured_output_model)
            self.expected_tool_name = self.structured_output_tool.tool_name

    @property
    def is_enabled(self) -> bool:
        """Check if structured output is enabled for this context.

        Returns:
            True if a structured output model is configured, False otherwise.
        """
        return self.structured_output_model is not None

    def store_result(self, tool_use_id: str, result: BaseModel) -> None:
        """Store a validated structured output result.

        Args:
            tool_use_id: Unique identifier for the tool use.
            result: Validated Pydantic model instance.
        """
        self.results[tool_use_id] = result

    def get_result(self, tool_use_id: str) -> BaseModel | None:
        """Retrieve a stored structured output result.

        Args:
            tool_use_id: Unique identifier for the tool use.

        Returns:
            The validated Pydantic model instance, or None if not found.
        """
        return self.results.get(tool_use_id)

    def set_forced_mode(self, tool_choice: dict | None = None) -> None:
        """Mark this context as being in forced structured output mode.

        Args:
            tool_choice: Optional tool choice configuration.
        """
        if not self.is_enabled:
            return
        self.forced_mode = True
        self.force_attempted = True
        self.tool_choice = tool_choice or {"any": {}}

    def has_structured_output_tool(self, tool_uses: list[ToolUse]) -> bool:
        """Check if any tool uses are for the structured output tool.

        Args:
            tool_uses: List of tool use dictionaries to check.

        Returns:
            True if any tool use matches the expected structured output tool name,
            False if no structured output tool is present or expected.
        """
        if not self.expected_tool_name:
            return False
        return any(tool_use.get("name") == self.expected_tool_name for tool_use in tool_uses)

    def get_tool_spec(self) -> Optional[ToolSpec]:
        """Get the tool specification for structured output.

        Returns:
            Tool specification, or None if no structured output model.
        """
        if self.structured_output_tool:
            return self.structured_output_tool.tool_spec
        return None

    def extract_result(self, tool_uses: list[ToolUse]) -> BaseModel | None:
        """Extract and remove structured output result from stored results.

        Args:
            tool_uses: List of tool use dictionaries from the current execution cycle.

        Returns:
            The structured output result if found, or None if no result available.
        """
        if not self.has_structured_output_tool(tool_uses):
            return None

        for tool_use in tool_uses:
            if tool_use.get("name") == self.expected_tool_name:
                tool_use_id = str(tool_use.get("toolUseId", ""))
                result = self.results.pop(tool_use_id, None)
                if result is not None:
                    logger.debug("Extracted structured output for %s", tool_use.get("name"))
                    return result
        return None

    def register_tool(self, registry: "ToolRegistry") -> None:
        """Register the structured output tool with the registry.

        Args:
            registry: The tool registry to register the tool with.
        """
        if self.structured_output_tool and self.structured_output_tool.tool_name not in registry.dynamic_tools:
            registry.register_dynamic_tool(self.structured_output_tool)
            logger.debug("Registered structured output tool: %s", self.structured_output_tool.tool_name)

    def cleanup(self, registry: "ToolRegistry") -> None:
        """Clean up the registered structured output tool from the registry.

        Args:
            registry: The tool registry to clean up the tool from.
        """
        if self.structured_output_tool and self.structured_output_tool.tool_name in registry.dynamic_tools:
            del registry.dynamic_tools[self.structured_output_tool.tool_name]
            logger.debug("Cleaned up structured output tool: %s", self.structured_output_tool.tool_name)

is_enabled property

Check if structured output is enabled for this context.

Returns:

Type Description
bool

True if a structured output model is configured, False otherwise.

__init__(structured_output_model=None)

Initialize a new structured output context.

Parameters:

Name Type Description Default
structured_output_model Type[BaseModel] | None

Optional Pydantic model type for structured output.

None
Source code in strands/tools/structured_output/_structured_output_context.py
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(self, structured_output_model: Type[BaseModel] | None = None):
    """Initialize a new structured output context.

    Args:
        structured_output_model: Optional Pydantic model type for structured output.
    """
    self.results: dict[str, BaseModel] = {}
    self.structured_output_model: Type[BaseModel] | None = structured_output_model
    self.structured_output_tool: StructuredOutputTool | None = None
    self.forced_mode: bool = False
    self.force_attempted: bool = False
    self.tool_choice: ToolChoice | None = None
    self.stop_loop: bool = False
    self.expected_tool_name: Optional[str] = None

    if structured_output_model:
        self.structured_output_tool = StructuredOutputTool(structured_output_model)
        self.expected_tool_name = self.structured_output_tool.tool_name

cleanup(registry)

Clean up the registered structured output tool from the registry.

Parameters:

Name Type Description Default
registry ToolRegistry

The tool registry to clean up the tool from.

required
Source code in strands/tools/structured_output/_structured_output_context.py
135
136
137
138
139
140
141
142
143
def cleanup(self, registry: "ToolRegistry") -> None:
    """Clean up the registered structured output tool from the registry.

    Args:
        registry: The tool registry to clean up the tool from.
    """
    if self.structured_output_tool and self.structured_output_tool.tool_name in registry.dynamic_tools:
        del registry.dynamic_tools[self.structured_output_tool.tool_name]
        logger.debug("Cleaned up structured output tool: %s", self.structured_output_tool.tool_name)

extract_result(tool_uses)

Extract and remove structured output result from stored results.

Parameters:

Name Type Description Default
tool_uses list[ToolUse]

List of tool use dictionaries from the current execution cycle.

required

Returns:

Type Description
BaseModel | None

The structured output result if found, or None if no result available.

Source code in strands/tools/structured_output/_structured_output_context.py
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def extract_result(self, tool_uses: list[ToolUse]) -> BaseModel | None:
    """Extract and remove structured output result from stored results.

    Args:
        tool_uses: List of tool use dictionaries from the current execution cycle.

    Returns:
        The structured output result if found, or None if no result available.
    """
    if not self.has_structured_output_tool(tool_uses):
        return None

    for tool_use in tool_uses:
        if tool_use.get("name") == self.expected_tool_name:
            tool_use_id = str(tool_use.get("toolUseId", ""))
            result = self.results.pop(tool_use_id, None)
            if result is not None:
                logger.debug("Extracted structured output for %s", tool_use.get("name"))
                return result
    return None

get_result(tool_use_id)

Retrieve a stored structured output result.

Parameters:

Name Type Description Default
tool_use_id str

Unique identifier for the tool use.

required

Returns:

Type Description
BaseModel | None

The validated Pydantic model instance, or None if not found.

Source code in strands/tools/structured_output/_structured_output_context.py
57
58
59
60
61
62
63
64
65
66
def get_result(self, tool_use_id: str) -> BaseModel | None:
    """Retrieve a stored structured output result.

    Args:
        tool_use_id: Unique identifier for the tool use.

    Returns:
        The validated Pydantic model instance, or None if not found.
    """
    return self.results.get(tool_use_id)

get_tool_spec()

Get the tool specification for structured output.

Returns:

Type Description
Optional[ToolSpec]

Tool specification, or None if no structured output model.

Source code in strands/tools/structured_output/_structured_output_context.py
 94
 95
 96
 97
 98
 99
100
101
102
def get_tool_spec(self) -> Optional[ToolSpec]:
    """Get the tool specification for structured output.

    Returns:
        Tool specification, or None if no structured output model.
    """
    if self.structured_output_tool:
        return self.structured_output_tool.tool_spec
    return None

has_structured_output_tool(tool_uses)

Check if any tool uses are for the structured output tool.

Parameters:

Name Type Description Default
tool_uses list[ToolUse]

List of tool use dictionaries to check.

required

Returns:

Type Description
bool

True if any tool use matches the expected structured output tool name,

bool

False if no structured output tool is present or expected.

Source code in strands/tools/structured_output/_structured_output_context.py
80
81
82
83
84
85
86
87
88
89
90
91
92
def has_structured_output_tool(self, tool_uses: list[ToolUse]) -> bool:
    """Check if any tool uses are for the structured output tool.

    Args:
        tool_uses: List of tool use dictionaries to check.

    Returns:
        True if any tool use matches the expected structured output tool name,
        False if no structured output tool is present or expected.
    """
    if not self.expected_tool_name:
        return False
    return any(tool_use.get("name") == self.expected_tool_name for tool_use in tool_uses)

register_tool(registry)

Register the structured output tool with the registry.

Parameters:

Name Type Description Default
registry ToolRegistry

The tool registry to register the tool with.

required
Source code in strands/tools/structured_output/_structured_output_context.py
125
126
127
128
129
130
131
132
133
def register_tool(self, registry: "ToolRegistry") -> None:
    """Register the structured output tool with the registry.

    Args:
        registry: The tool registry to register the tool with.
    """
    if self.structured_output_tool and self.structured_output_tool.tool_name not in registry.dynamic_tools:
        registry.register_dynamic_tool(self.structured_output_tool)
        logger.debug("Registered structured output tool: %s", self.structured_output_tool.tool_name)

set_forced_mode(tool_choice=None)

Mark this context as being in forced structured output mode.

Parameters:

Name Type Description Default
tool_choice dict | None

Optional tool choice configuration.

None
Source code in strands/tools/structured_output/_structured_output_context.py
68
69
70
71
72
73
74
75
76
77
78
def set_forced_mode(self, tool_choice: dict | None = None) -> None:
    """Mark this context as being in forced structured output mode.

    Args:
        tool_choice: Optional tool choice configuration.
    """
    if not self.is_enabled:
        return
    self.forced_mode = True
    self.force_attempted = True
    self.tool_choice = tool_choice or {"any": {}}

store_result(tool_use_id, result)

Store a validated structured output result.

Parameters:

Name Type Description Default
tool_use_id str

Unique identifier for the tool use.

required
result BaseModel

Validated Pydantic model instance.

required
Source code in strands/tools/structured_output/_structured_output_context.py
48
49
50
51
52
53
54
55
def store_result(self, tool_use_id: str, result: BaseModel) -> None:
    """Store a validated structured output result.

    Args:
        tool_use_id: Unique identifier for the tool use.
        result: Validated Pydantic model instance.
    """
    self.results[tool_use_id] = result

StructuredOutputTool

Bases: AgentTool

Tool implementation for structured output validation.

Source code in strands/tools/structured_output/structured_output_tool.py
 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
class StructuredOutputTool(AgentTool):
    """Tool implementation for structured output validation."""

    def __init__(self, structured_output_model: Type[BaseModel]) -> None:
        """Initialize a structured output tool.

        Args:
            structured_output_model: The Pydantic model class that defines the expected output structure.
        """
        super().__init__()
        self._structured_output_type = structured_output_model
        self._tool_spec = self._get_tool_spec(structured_output_model)
        self._tool_spec["description"] = (
            "IMPORTANT: This StructuredOutputTool should only be invoked as the last and final tool "
            f"before returning the completed result to the caller. "
            f"<description>{self._tool_spec.get('description', '')}</description>"
        )
        self._tool_name = self._tool_spec.get("name", "StructuredOutputTool")

    @classmethod
    def _get_tool_spec(cls, structured_output_model: Type[BaseModel]) -> ToolSpec:
        """Get a cached tool spec for the given output type.

        Args:
            structured_output_model: The Pydantic model class that defines the expected output structure.

        Returns:
            Cached tool specification for the output type.
        """
        if structured_output_model not in _TOOL_SPEC_CACHE:
            _TOOL_SPEC_CACHE[structured_output_model] = convert_pydantic_to_tool_spec(structured_output_model)
        return deepcopy(_TOOL_SPEC_CACHE[structured_output_model])

    @property
    def tool_name(self) -> str:
        """Get the name of the tool.

        Returns:
            The name of the tool (same as the Pydantic model class name).
        """
        return self._tool_name

    @property
    def tool_spec(self) -> ToolSpec:
        """Get the tool specification for this structured output tool.

        Returns:
            The tool specification generated from the Pydantic model.
        """
        return self._tool_spec

    @property
    def tool_type(self) -> str:
        """Identifies this as a structured output tool implementation.

        Returns:
            "structured_output".
        """
        return "structured_output"

    @property
    def structured_output_model(self) -> Type[BaseModel]:
        """Get the Pydantic model type for this tool.

        Returns:
            The Pydantic model class.
        """
        return self._structured_output_type

    @override
    async def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
        """Validate the structured output and return appropriate result.

        Args:
            tool_use: The tool use request containing the data to validate.
            invocation_state: Context for the tool invocation (kept for compatibility).
            **kwargs: Additional keyword arguments, including structured_output_context.

        Yields:
            Tool events with the last being the tool result (success or error).
        """
        tool_input: dict[str, Any] = tool_use.get("input", {})
        tool_use_id = str(tool_use.get("toolUseId", ""))

        context: StructuredOutputContext = kwargs.get("structured_output_context")  # type: ignore
        try:
            validated_object = self._structured_output_type(**tool_input)
            logger.debug("tool_name=<%s> | structured output validated", self._tool_name)
            context.store_result(tool_use_id, validated_object)

            result: ToolResult = {
                "toolUseId": tool_use_id,
                "status": "success",
                "content": [{"text": f"Successfully validated {self._tool_name} structured output"}],
            }

            yield ToolResultEvent(result)

        except ValidationError as e:
            error_details = []
            for error in e.errors():
                field_path = " -> ".join(str(loc) for loc in error["loc"]) if error["loc"] else "root"
                error_details.append(f"Field '{field_path}': {error['msg']}")

            error_message = f"Validation failed for {self._tool_name}. Please fix the following errors:\n" + "\n".join(
                f"- {detail}" for detail in error_details
            )
            logger.error(
                "tool_name=<%s> | structured output validation failed | error_message=<%s>",
                self._tool_name,
                error_message,
            )

            # Create error result that will be sent back to the LLM so it can decide if it needs to retry
            validation_error_result: ToolResult = {
                "toolUseId": tool_use_id,
                "status": "error",
                "content": [{"text": error_message}],
            }

            yield ToolResultEvent(validation_error_result)

        except Exception as e:
            error_message = f"Unexpected error validating {self._tool_name}: {str(e)}"
            logger.exception(error_message)

            exception_result: ToolResult = {
                "toolUseId": tool_use_id,
                "status": "error",
                "content": [{"text": error_message}],
            }

            yield ToolResultEvent(exception_result)

structured_output_model property

Get the Pydantic model type for this tool.

Returns:

Type Description
Type[BaseModel]

The Pydantic model class.

tool_name property

Get the name of the tool.

Returns:

Type Description
str

The name of the tool (same as the Pydantic model class name).

tool_spec property

Get the tool specification for this structured output tool.

Returns:

Type Description
ToolSpec

The tool specification generated from the Pydantic model.

tool_type property

Identifies this as a structured output tool implementation.

Returns:

Type Description
str

"structured_output".

__init__(structured_output_model)

Initialize a structured output tool.

Parameters:

Name Type Description Default
structured_output_model Type[BaseModel]

The Pydantic model class that defines the expected output structure.

required
Source code in strands/tools/structured_output/structured_output_tool.py
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, structured_output_model: Type[BaseModel]) -> None:
    """Initialize a structured output tool.

    Args:
        structured_output_model: The Pydantic model class that defines the expected output structure.
    """
    super().__init__()
    self._structured_output_type = structured_output_model
    self._tool_spec = self._get_tool_spec(structured_output_model)
    self._tool_spec["description"] = (
        "IMPORTANT: This StructuredOutputTool should only be invoked as the last and final tool "
        f"before returning the completed result to the caller. "
        f"<description>{self._tool_spec.get('description', '')}</description>"
    )
    self._tool_name = self._tool_spec.get("name", "StructuredOutputTool")

stream(tool_use, invocation_state, **kwargs) async

Validate the structured output and return appropriate result.

Parameters:

Name Type Description Default
tool_use ToolUse

The tool use request containing the data to validate.

required
invocation_state dict[str, Any]

Context for the tool invocation (kept for compatibility).

required
**kwargs Any

Additional keyword arguments, including structured_output_context.

{}

Yields:

Type Description
ToolGenerator

Tool events with the last being the tool result (success or error).

Source code in strands/tools/structured_output/structured_output_tool.py
 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
@override
async def stream(self, tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> ToolGenerator:
    """Validate the structured output and return appropriate result.

    Args:
        tool_use: The tool use request containing the data to validate.
        invocation_state: Context for the tool invocation (kept for compatibility).
        **kwargs: Additional keyword arguments, including structured_output_context.

    Yields:
        Tool events with the last being the tool result (success or error).
    """
    tool_input: dict[str, Any] = tool_use.get("input", {})
    tool_use_id = str(tool_use.get("toolUseId", ""))

    context: StructuredOutputContext = kwargs.get("structured_output_context")  # type: ignore
    try:
        validated_object = self._structured_output_type(**tool_input)
        logger.debug("tool_name=<%s> | structured output validated", self._tool_name)
        context.store_result(tool_use_id, validated_object)

        result: ToolResult = {
            "toolUseId": tool_use_id,
            "status": "success",
            "content": [{"text": f"Successfully validated {self._tool_name} structured output"}],
        }

        yield ToolResultEvent(result)

    except ValidationError as e:
        error_details = []
        for error in e.errors():
            field_path = " -> ".join(str(loc) for loc in error["loc"]) if error["loc"] else "root"
            error_details.append(f"Field '{field_path}': {error['msg']}")

        error_message = f"Validation failed for {self._tool_name}. Please fix the following errors:\n" + "\n".join(
            f"- {detail}" for detail in error_details
        )
        logger.error(
            "tool_name=<%s> | structured output validation failed | error_message=<%s>",
            self._tool_name,
            error_message,
        )

        # Create error result that will be sent back to the LLM so it can decide if it needs to retry
        validation_error_result: ToolResult = {
            "toolUseId": tool_use_id,
            "status": "error",
            "content": [{"text": error_message}],
        }

        yield ToolResultEvent(validation_error_result)

    except Exception as e:
        error_message = f"Unexpected error validating {self._tool_name}: {str(e)}"
        logger.exception(error_message)

        exception_result: ToolResult = {
            "toolUseId": tool_use_id,
            "status": "error",
            "content": [{"text": error_message}],
        }

        yield ToolResultEvent(exception_result)

ToolResult

Bases: TypedDict

Result of a tool execution.

Attributes:

Name Type Description
content list[ToolResultContent]

List of result content returned by the tool.

status ToolResultStatus

The status of the tool execution ("success" or "error").

toolUseId str

The unique identifier of the tool use request that produced this result.

Source code in strands/types/tools.py
87
88
89
90
91
92
93
94
95
96
97
98
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

ToolResultEvent

Bases: TypedEvent

Event emitted when a tool execution completes.

Source code in strands/types/_events.py
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
class ToolResultEvent(TypedEvent):
    """Event emitted when a tool execution completes."""

    def __init__(self, tool_result: ToolResult) -> None:
        """Initialize with the completed tool result.

        Args:
            tool_result: Final result from the tool execution
        """
        super().__init__({"type": "tool_result", "tool_result": tool_result})

    @property
    def tool_use_id(self) -> str:
        """The toolUseId associated with this result."""
        return cast(ToolResult, self.get("tool_result"))["toolUseId"]

    @property
    def tool_result(self) -> ToolResult:
        """Final result from the completed tool execution."""
        return cast(ToolResult, self.get("tool_result"))

    @property
    @override
    def is_callback_event(self) -> bool:
        return False

tool_result property

Final result from the completed tool execution.

tool_use_id property

The toolUseId associated with this result.

__init__(tool_result)

Initialize with the completed tool result.

Parameters:

Name Type Description Default
tool_result ToolResult

Final result from the tool execution

required
Source code in strands/types/_events.py
278
279
280
281
282
283
284
def __init__(self, tool_result: ToolResult) -> None:
    """Initialize with the completed tool result.

    Args:
        tool_result: Final result from the tool execution
    """
    super().__init__({"type": "tool_result", "tool_result": tool_result})

ToolSpec

Bases: TypedDict

Specification for a tool that can be used by an agent.

Attributes:

Name Type Description
description str

A human-readable description of what the tool does.

inputSchema JSONSchema

JSON Schema defining the expected input parameters.

name str

The unique name of the tool.

outputSchema NotRequired[JSONSchema]

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.

Source code in strands/types/tools.py
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
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]

ToolUse

Bases: TypedDict

A request from the model to use a specific tool with the provided input.

Attributes:

Name Type Description
input Any

The input parameters for the tool. Can be any JSON-serializable type.

name str

The name of the tool to invoke.

toolUseId str

A unique identifier for this specific tool use request.

Source code in strands/types/tools.py
52
53
54
55
56
57
58
59
60
61
62
63
64
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

convert_pydantic_to_tool_spec(model, description=None)

Converts a Pydantic model to a tool description for the Amazon Bedrock Converse API.

Handles optional vs. required fields, resolves $refs, and uses docstrings.

Parameters:

Name Type Description Default
model Type[BaseModel]

The Pydantic model class to convert

required
description Optional[str]

Optional description of the tool's purpose

None

Returns:

Name Type Description
ToolSpec ToolSpec

Dict containing the Bedrock tool specification

Source code in strands/tools/structured_output/structured_output_utils.py
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
def convert_pydantic_to_tool_spec(
    model: Type[BaseModel],
    description: Optional[str] = None,
) -> ToolSpec:
    """Converts a Pydantic model to a tool description for the Amazon Bedrock Converse API.

    Handles optional vs. required fields, resolves $refs, and uses docstrings.

    Args:
        model: The Pydantic model class to convert
        description: Optional description of the tool's purpose

    Returns:
        ToolSpec: Dict containing the Bedrock tool specification
    """
    name = model.__name__

    # Get the JSON schema
    input_schema = model.model_json_schema()

    # Get model docstring for description if not provided
    model_description = description
    if not model_description and model.__doc__:
        model_description = model.__doc__.strip()

    # Process all referenced models to ensure proper docstrings
    # This step is important for gathering descriptions from referenced models
    _process_referenced_models(input_schema, model)

    # Now, let's fully expand the nested models with all their properties
    _expand_nested_properties(input_schema, model)

    # Flatten the schema
    flattened_schema = _flatten_schema(input_schema)

    final_schema = flattened_schema

    # Construct the tool specification
    return ToolSpec(
        name=name,
        description=model_description or f"{name} structured output tool",
        inputSchema={"json": final_schema},
    )