Skip to content

strands.handlers

Various handlers for performing custom actions on agent state.

Examples include:

  • Processing tool invocations
  • Displaying events from the event stream

strands.handlers.callback_handler

This module provides handlers for formatting and displaying events from the agent.

CompositeCallbackHandler

Class-based callback handler that combines multiple callback handlers.

This handler allows multiple callback handlers to be invoked for the same events, enabling different processing or output formats for the same stream data.

Source code in strands/handlers/callback_handler.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class CompositeCallbackHandler:
    """Class-based callback handler that combines multiple callback handlers.

    This handler allows multiple callback handlers to be invoked for the same events,
    enabling different processing or output formats for the same stream data.
    """

    def __init__(self, *handlers: Callable) -> None:
        """Initialize handler."""
        self.handlers = handlers

    def __call__(self, **kwargs: Any) -> None:
        """Invoke all handlers in the chain."""
        for handler in self.handlers:
            handler(**kwargs)

__call__(**kwargs)

Invoke all handlers in the chain.

Source code in strands/handlers/callback_handler.py 
58
59
60
61
def __call__(self, **kwargs: Any) -> None:
    """Invoke all handlers in the chain."""
    for handler in self.handlers:
        handler(**kwargs)

__init__(*handlers)

Initialize handler.

Source code in strands/handlers/callback_handler.py 
54
55
56
def __init__(self, *handlers: Callable) -> None:
    """Initialize handler."""
    self.handlers = handlers

PrintingCallbackHandler

Handler for streaming text output and tool invocations to stdout.

Source code in strands/handlers/callback_handler.py 
 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
class PrintingCallbackHandler:
    """Handler for streaming text output and tool invocations to stdout."""

    def __init__(self) -> None:
        """Initialize handler."""
        self.tool_count = 0
        self.previous_tool_use = None

    def __call__(self, **kwargs: Any) -> None:
        """Stream text output and tool invocations to stdout.

        Args:
            **kwargs: Callback event data including:
                - reasoningText (Optional[str]): Reasoning text to print if provided.
                - data (str): Text content to stream.
                - complete (bool): Whether this is the final chunk of a response.
                - current_tool_use (dict): Information about the current tool being used.
        """
        reasoningText = kwargs.get("reasoningText", False)
        data = kwargs.get("data", "")
        complete = kwargs.get("complete", False)
        current_tool_use = kwargs.get("current_tool_use", {})

        if reasoningText:
            print(reasoningText, end="")

        if data:
            print(data, end="" if not complete else "\n")

        if current_tool_use and current_tool_use.get("name"):
            tool_name = current_tool_use.get("name", "Unknown tool")
            if self.previous_tool_use != current_tool_use:
                self.previous_tool_use = current_tool_use
                self.tool_count += 1
                print(f"\nTool #{self.tool_count}: {tool_name}")

        if complete and data:
            print("\n")

__call__(**kwargs)

Stream text output and tool invocations to stdout.

Parameters:

Name Type Description Default
**kwargs Any

Callback event data including: - reasoningText (Optional[str]): Reasoning text to print if provided. - data (str): Text content to stream. - complete (bool): Whether this is the final chunk of a response. - current_tool_use (dict): Information about the current tool being used.

 {}
Source code in strands/handlers/callback_handler.py 
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
def __call__(self, **kwargs: Any) -> None:
    """Stream text output and tool invocations to stdout.

    Args:
        **kwargs: Callback event data including:
            - reasoningText (Optional[str]): Reasoning text to print if provided.
            - data (str): Text content to stream.
            - complete (bool): Whether this is the final chunk of a response.
            - current_tool_use (dict): Information about the current tool being used.
    """
    reasoningText = kwargs.get("reasoningText", False)
    data = kwargs.get("data", "")
    complete = kwargs.get("complete", False)
    current_tool_use = kwargs.get("current_tool_use", {})

    if reasoningText:
        print(reasoningText, end="")

    if data:
        print(data, end="" if not complete else "\n")

    if current_tool_use and current_tool_use.get("name"):
        tool_name = current_tool_use.get("name", "Unknown tool")
        if self.previous_tool_use != current_tool_use:
            self.previous_tool_use = current_tool_use
            self.tool_count += 1
            print(f"\nTool #{self.tool_count}: {tool_name}")

    if complete and data:
        print("\n")

__init__()

Initialize handler.

Source code in strands/handlers/callback_handler.py 
10
11
12
13
def __init__(self) -> None:
    """Initialize handler."""
    self.tool_count = 0
    self.previous_tool_use = None

null_callback_handler(**_kwargs)

Callback handler that discards all output.

Parameters:

Name Type Description Default
**_kwargs Any

Event data (ignored).

 {}
Source code in strands/handlers/callback_handler.py 
64
65
66
67
68
69
70
def null_callback_handler(**_kwargs: Any) -> None:
    """Callback handler that discards all output.

    Args:
        **_kwargs: Event data (ignored).
    """
    return None

strands.handlers.tool_handler

This module provides handlers for managing tool invocations.

AgentToolHandler

Bases: ToolHandler

Handler for processing tool invocations in agent.

This class implements the ToolHandler interface and provides functionality for looking up tools in a registry and invoking them with the appropriate parameters.

Source code in strands/handlers/tool_handler.py 
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
class AgentToolHandler(ToolHandler):
    """Handler for processing tool invocations in agent.

    This class implements the ToolHandler interface and provides functionality for looking up tools in a registry and
    invoking them with the appropriate parameters.
    """

    def __init__(self, tool_registry: ToolRegistry) -> None:
        """Initialize handler.

        Args:
            tool_registry: Registry of available tools.
        """
        self.tool_registry = tool_registry

    def process(
        self,
        tool: ToolUse,
        *,
        model: Model,
        system_prompt: Optional[str],
        messages: Messages,
        tool_config: ToolConfig,
        kwargs: dict[str, Any],
    ) -> ToolGenerator:
        """Process a tool invocation.

        Looks up the tool in the registry and invokes it with the provided parameters.

        Args:
            tool: The tool object to process, containing name and parameters.
            model: The model being used for the agent.
            system_prompt: The system prompt for the agent.
            messages: The conversation history.
            tool_config: Configuration for the tool.
            kwargs: Additional keyword arguments passed to the tool.

        Yields:
            Events of the tool invocation.

        Returns:
            The final tool result or an error response if the tool fails or is not found.
        """
        logger.debug("tool=<%s> | invoking", tool)
        tool_use_id = tool["toolUseId"]
        tool_name = tool["name"]

        # Get the tool info
        tool_info = self.tool_registry.dynamic_tools.get(tool_name)
        tool_func = tool_info if tool_info is not None else self.tool_registry.registry.get(tool_name)

        try:
            # Check if tool exists
            if not tool_func:
                logger.error(
                    "tool_name=<%s>, available_tools=<%s> | tool not found in registry",
                    tool_name,
                    list(self.tool_registry.registry.keys()),
                )
                return {
                    "toolUseId": tool_use_id,
                    "status": "error",
                    "content": [{"text": f"Unknown tool: {tool_name}"}],
                }
            # Add standard arguments to kwargs for Python tools
            kwargs.update(
                {
                    "model": model,
                    "system_prompt": system_prompt,
                    "messages": messages,
                    "tool_config": tool_config,
                }
            )

            result = tool_func.invoke(tool, **kwargs)
            yield {"result": result}  # Placeholder until tool_func becomes a generator from which we can yield from
            return result

        except Exception as e:
            logger.exception("tool_name=<%s> | failed to process tool", tool_name)
            return {
                "toolUseId": tool_use_id,
                "status": "error",
                "content": [{"text": f"Error: {str(e)}"}],
            }

__init__(tool_registry)

Initialize handler.

Parameters:

Name Type Description Default
tool_registry ToolRegistry

Registry of available tools.

 required
Source code in strands/handlers/tool_handler.py 
21
22
23
24
25
26
27
def __init__(self, tool_registry: ToolRegistry) -> None:
    """Initialize handler.

    Args:
        tool_registry: Registry of available tools.
    """
    self.tool_registry = tool_registry

process(tool, *, model, system_prompt, messages, tool_config, kwargs)

Process a tool invocation.

Looks up the tool in the registry and invokes it with the provided parameters.

Parameters:

Name Type Description Default
tool ToolUse

The tool object to process, containing name and parameters.

 required
model Model

The model being used for the agent.

 required
system_prompt Optional[str]

The system prompt for the agent.

 required
messages Messages

The conversation history.

 required
tool_config ToolConfig

Configuration for the tool.

 required
kwargs dict[str, Any]

Additional keyword arguments passed to the tool.

 required

Yields:

Type Description
ToolGenerator

Events of the tool invocation.

Returns:

Type Description
ToolGenerator

The final tool result or an error response if the tool fails or is not found.
Source code in strands/handlers/tool_handler.py 
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
def process(
    self,
    tool: ToolUse,
    *,
    model: Model,
    system_prompt: Optional[str],
    messages: Messages,
    tool_config: ToolConfig,
    kwargs: dict[str, Any],
) -> ToolGenerator:
    """Process a tool invocation.

    Looks up the tool in the registry and invokes it with the provided parameters.

    Args:
        tool: The tool object to process, containing name and parameters.
        model: The model being used for the agent.
        system_prompt: The system prompt for the agent.
        messages: The conversation history.
        tool_config: Configuration for the tool.
        kwargs: Additional keyword arguments passed to the tool.

    Yields:
        Events of the tool invocation.

    Returns:
        The final tool result or an error response if the tool fails or is not found.
    """
    logger.debug("tool=<%s> | invoking", tool)
    tool_use_id = tool["toolUseId"]
    tool_name = tool["name"]

    # Get the tool info
    tool_info = self.tool_registry.dynamic_tools.get(tool_name)
    tool_func = tool_info if tool_info is not None else self.tool_registry.registry.get(tool_name)

    try:
        # Check if tool exists
        if not tool_func:
            logger.error(
                "tool_name=<%s>, available_tools=<%s> | tool not found in registry",
                tool_name,
                list(self.tool_registry.registry.keys()),
            )
            return {
                "toolUseId": tool_use_id,
                "status": "error",
                "content": [{"text": f"Unknown tool: {tool_name}"}],
            }
        # Add standard arguments to kwargs for Python tools
        kwargs.update(
            {
                "model": model,
                "system_prompt": system_prompt,
                "messages": messages,
                "tool_config": tool_config,
            }
        )

        result = tool_func.invoke(tool, **kwargs)
        yield {"result": result}  # Placeholder until tool_func becomes a generator from which we can yield from
        return result

    except Exception as e:
        logger.exception("tool_name=<%s> | failed to process tool", tool_name)
        return {
            "toolUseId": tool_use_id,
            "status": "error",
            "content": [{"text": f"Error: {str(e)}"}],
        }