github strands-agents/sdk-python v1.20.0

2 days ago

Major Features

Swarm Interrupts for Human-in-the-Loop - PR#1193

Swarm multi-agent systems now support interrupts, enabling Human-in-the-Loop patterns for approval workflows and user interaction during agent execution. Interrupts can be triggered via BeforeNodeCallEvent hooks or directly within agent tools using ToolContext.

from strands import Agent, tool
from strands.experimental.hooks.multiagent import BeforeNodeCallEvent
from strands.hooks import HookProvider
from strands.multiagent import Swarm
from strands.multiagent.base import Status

# Example 1: Interrupt via Hook
class ApprovalHook(HookProvider):
    def register_hooks(self, registry):
        registry.add_callback(BeforeNodeCallEvent, self.approve)

    def approve(self, event):
        response = event.interrupt("approval", reason=f"{event.node_id} needs approval")
        if response != "APPROVE":
            event.cancel_node = "rejected"

swarm = Swarm([agent1, agent2], hooks=[ApprovalHook()])
result = swarm("Task requiring approval")

# Handle interrupts
while result.status == Status.INTERRUPTED:
    for interrupt in result.interrupts:
        user_input = input(f"{interrupt.reason}: ")
        responses = [{"interruptResponse": {"interruptId": interrupt.id, "response": user_input}}]
    result = swarm(responses)

# Example 2: Interrupt via Tool
@tool(context=True)
def get_user_info(tool_context: ToolContext) -> str:
    response = tool_context.interrupt("user_info", reason="need user name")
    return f"User: {response}"

user_agent = Agent(name="user", tools=[get_user_info])
swarm = Swarm([user_agent])
result = swarm("Who is the user?")
# Resume with interrupt response as shown above

See the interrupts documentation for more details.

AgentResult Access in AfterInvocationEvent - PR#1125

Hooks can now access the complete AgentResult in AfterInvocationEvent, enabling post-invocation actions based on the agent's output, stop reason, and metrics. This enhancement allows for richer observability and custom handling of agent results.

from strands import Agent
from strands.hooks import AfterInvocationEvent, HookProvider

class ResultLoggingHook(HookProvider):
    def register_hooks(self, registry):
        registry.add_callback(AfterInvocationEvent, self.log_result)

    def log_result(self, event: AfterInvocationEvent):
        # Access the complete AgentResult
        if event.result:
            print(f"Stop reason: {event.result.stop_reason}")
            print(f"Tokens used: {event.result.usage}")
            print(f"Response: {event.result.text}")

agent = Agent(hooks=[ResultLoggingHook()])
result = agent("What is 2+2?")

Major Bug Fixes

  • Structured Output Display Fix - PR#1290
    Fixed AgentResult.__str__() to return structured output JSON when no text content is present, resolving issues where print(agent_result) showed empty output and structured output was lost in multi-agent graph propagation.

  • Tool Spec Composition Keywords Fix - PR#1301
    Fixed tool specification handling for JSON Schema composition keywords (anyOf, oneOf, allOf, not), preventing models from incorrectly returning string-encoded JSON for optional parameters like Optional[List[str]].

  • MCP Client Resource Leak Fix - PR#1321
    Fixed file descriptor leak in MCP client by properly closing the asyncio event loop, preventing resource exhaustion in multi-tenant applications that create many MCP clients.


All Changes

New Contributors

Full Changelog: v1.19.0...v1.20.0

Don't miss a new sdk-python release

NewReleases is sending notifications on new releases.