[FEATURE] Support for Direct Method Call Streaming

[larrylp2]

Problem Statement

Strands supports direct method calls (https://strandsagents.com/latest/documentation/docs/user-guide/concepts/tools/#direct-method-calls), which enables developers to invoke specific tools directly.

However, direct method calls currently do not support streaming results for developers. This means that there is a disjointed experience for the same Strands agent, where stream_async enables streaming responses to users for E2E Agent calls but are not possible for direct method calls.

Proposed Solution

To address this, the tools/_caller.py can be extended to also have an async version.

The current direct tool call method seems to use a streaming-friendly ToolExecutor call behind the scenes, but handles it synchronously before returning the full tool result.

https://github.com/strands-agents/sdk-python/blob/main/src/strands/tools/_caller.py#L89

Use Case

The use case would be a seamless streaming experience, even if an Agent built with Strands incorporates direct tool calls in certain flows behind the scenes.

A multi-agent app following the Agents as Tools (https://strandsagents.com/latest/documentation/docs/user-guide/concepts/multi-agent/agents-as-tools/) pattern that relies on streamed tool events (https://strandsagents.com/latest/documentation/docs/user-guide/concepts/streaming/#tool-events) can still stream results through direct method calls.

Alternatives Solutions

No response

Additional Context

No response

[cagataycali]

Hi @larrylp2! Great feature request - streaming support for direct method calls would definitely improve the developer experience for multi-agent patterns.

Feedback on Proposed Solution:

Your observation about ToolExecutor already using streaming-friendly calls is spot on! The current implementation does seem to have the infrastructure in place. A few thoughts:

1. API Design - Would you envision something like:

   async for chunk in agent.tool.my_tool.stream_async():
       # Handle streamed tool events
       if chunk.type == "tool_output":
           print(chunk.content)

   Or would it follow the existing stream_async() pattern more closely with ToolUseEvent, ToolResultEvent, etc.?

2. Event Types - Should direct method call streaming emit the same event types as agent-level streaming (ToolUseEvent, ToolResultEvent), or a simplified subset since we're not dealing with agent reasoning/planning?

3. Error Handling - How should streaming errors be surfaced? Should they follow the same pattern as agent.stream_async() or have direct-call-specific handling?

Additional Considerations:

4. Sync vs Async - Would you also want a synchronous streaming variant (generator-based), or is async sufficient for your use case?

5. Backward Compatibility - The current direct call syntax (agent.tool.tool_name()) is clean. Should streaming be:

   • A separate method: agent.tool.tool_name.stream_async()
   • A parameter: agent.tool.tool_name(stream=True)
   • Something else?

6. Multi-Agent Streaming - For the Agents-as-Tools pattern you mentioned, would you need to distinguish between events from different agent layers, or is a flat stream acceptable?

This would be a valuable addition for building responsive multi-agent systems! Happy to help with design discussions or implementation. 🦆

[zastrowm]

I'm inclined - I would expect new methods off of the tool caller that can stream results. Both synchronous and asynchronous method calls probably. I would expect some design work into covering all use cases, but otherwise aligned.

I think we can open this up to contribution even with the additional refinement needed

[emaan-c]

Direct Tool Call Streaming API

Status: Proposed

Date: 2026-03-22

Issue: #1436

Context

Direct tool calls (agent.tool.my_tool()) currently block without providing streaming events, while agent-level calls (agent.stream_async()) expose rich streaming events from the same ToolExecutor._stream() infrastructure. This creates inconsistent developer experience and prevents building responsive UIs for long-running operations, multi-agent systems, and debugging workflows.

Who experiences this: Developers building multi-agent systems (Agents-as-Tools pattern), applications with long-running tools (database queries, API calls), and responsive UIs requiring real-time progress feedback.

Decision

Add streaming methods to tool call objects by returning a callable _ToolExecutor wrapper instead of raw functions from _ToolCaller.__getattr__:

class _ToolExecutor:
    def __call__(self, **kwargs) -> ToolResult:
        """Sync execution (existing behavior - backward compatible)"""
    
    def stream(self, **kwargs) -> Iterator[TypedEvent]:
        """Sync streaming (NEW)"""
    
    async def stream_async(self, **kwargs) -> AsyncIterator[TypedEvent]:
        """Async streaming (NEW)"""

Why this approach:

1. Matches @zastrowm feedback: "new methods off of the tool caller" - not a separate property
2. Addresses full requirement: Both sync and async streaming as requested
3. Pattern consistency: Mirrors existing Agent pattern (agent() + agent.stream_async())
4. Type safe: Each method has distinct, clear return type

Developer Experience

Basic usage:

# Backward compatible sync (unchanged)
result = agent.tool.fetch_data(query="users")

# NEW: Sync streaming
for event in agent.tool.fetch_data.stream(query="users"):
    print(f"Progress: {event.get('type')}")

# NEW: Async streaming
async for event in agent.tool.fetch_data.stream_async(query="users"):
    await ui.update_progress(event)

Multi-agent composition:

research_agent = Agent(tools=[database_search, web_scrape])
main_agent = Agent(tools=[research_agent])

# Stream sub-agent execution with full visibility
async for event in main_agent.tool.research_agent.stream_async(prompt="Research X"):
    if event.get("data"):
        print(f"Research: {event['data']}")

Error handling: Standard Python async error handling - exceptions propagate naturally.

Alternatives Considered

1. Property-based: agent.tool_async.my_tool()

• Rejected: @zastrowm explicitly requested methods on tool caller, not separate property. Only provided async, missed sync requirement.

2. Parameter-based: agent.tool.my_tool(stream=True)

• Rejected: Return type varies by parameter (poor type safety), doesn't match Agent pattern.

3. Suffix convention: agent.tool.my_tool_stream()

• Rejected: Pollutes namespace with shadow methods, inconsistent with agent.stream_async() naming.

Why methods win: Matches existing Agent pattern exactly, type-safe, discoverable via IDE autocomplete, explicitly separates execution modes.

Consequences

Easier:

• Building responsive UIs with real-time progress
• Multi-agent systems with visibility at all levels
• Debugging tool execution with observability
• Consistent streaming mental model across agent and tool calls

More difficult:

• Slightly larger API surface (adds 2 methods per tool)
• Documentation scope increases (3 execution modes instead of 1)

Breaking changes: None. Existing agent.tool.my_tool() behavior unchanged. New methods are opt-in additions.

Willingness to Implement

Yes - Have already published a PR for this and awaiting review.