feat: add streaming to direct tool calls

[emaan-c]

Description

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

This adds stream() and stream_async() methods to tool calls, enabling real-time observability without recording to message history.

Resolves: #1436

Public API Changes

Tool calls now support three execution modes:

# Synchronous (unchanged - backward compatible)
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)

Streaming methods yield the same events as ToolExecutor._stream() without recording to message history. 100% backward compatible.

Use Cases

• Multi-agent composition: Stream sub-agent execution with full visibility
• Real-time progress: Display live indicators during long-running operations
• Debugging: Inspect execution flow without polluting conversation history

Related Issues

#1436

Documentation PR

No documentation PR needed.

Type of Change

New feature

Testing

• I ran hatch run prepare

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[codecov]

Codecov Report

❌ Patch coverage is 98.64865% with 1 line in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/strands/tools/_caller.py	98.64%	0 Missing and 1 partial ⚠️

📢 Thoughts on this report? Let us know!

[mkmeral]

/strands review

[mkmeral]

do we need a separate Tool Executor, why not just add stream methods to toolcaller?

Also strands already has tool executor concept, so there is a name clash

[mkmeral]

is this path separate from __call__? Can we make streaming default and return final event (like how we handle agent stream and call)

[github-actions]

Issue: stream() collects all events into a list before yielding, which defeats the purpose of streaming. Users calling stream() will not see events in real-time — they'll get all events at once after tool execution completes, making this behave identically to __call__() but with extra overhead.

Suggestion: Consider using a thread-safe queue to bridge async-to-sync streaming, or at minimum, make the docstring and PR description very explicit that stream() does NOT provide real-time streaming. Currently the docstring says "events are buffered before yielding" but the PR description positions this as enabling "real-time progress" which is misleading for the sync variant.

An alternative approach using a queue:

def stream(self, **kwargs: Any) -> Iterator[TypedEvent]:
    import queue
    import threading
    
    q: queue.Queue[TypedEvent | None] = queue.Queue()
    
    async def _produce() -> None:
        try:
            async for event in self.stream_async(**kwargs):
                q.put(event)
        finally:
            q.put(None)  # sentinel
    
    thread = threading.Thread(target=lambda: run_async(_produce), daemon=True)
    thread.start()
    
    while True:
        item = q.get()
        if item is None:
            break
        yield item
    thread.join()

If a true sync streaming implementation is too complex for this PR, consider removing stream() entirely and only shipping stream_async(). A sync method that doesn't actually stream could confuse users (violates "the obvious path is the happy path" tenet).

[github-actions]

Issue: stream_async() doesn't check the invocation lock or record_direct_tool_call, while __call__() does. This means streaming tool calls could execute concurrently with agent invocations even when record_direct_tool_call is True (the default), creating potential race conditions with shared agent state.

Suggestion: Either add the same concurrency guard from __call__(), or document why streaming calls intentionally skip it (e.g., because they don't record to message history). If the intent is that streaming never records, make this an explicit design decision in the docstring.

[github-actions]

Issue: The return type AsyncIterator[TypedEvent] for an async def with yield should be AsyncGenerator[TypedEvent, None] or AsyncIterator[TypedEvent]. However, TypedEvent is from the private _events module and is not exported from strands or strands.types. Customers wanting to type-hint the events they receive from this public API have no stable import path.

Suggestion: Either export TypedEvent from a public module (e.g., strands.types or strands), or use a more generic type like dict[str, Any] for the public API signature until TypedEvent is promoted to a public type.

[github-actions]

Issue: The tool_use setup logic (normalize name, create tool_id, build ToolUse dict, create tool_results list) is duplicated between __call__() and stream_async(). This violates DRY and creates a maintenance burden — any future change to tool setup must be applied in both places.

Suggestion: Extract a private helper method, e.g.:

def _prepare_tool_use(self, **kwargs: Any) -> tuple[ToolUse, list[ToolResult], dict]:
    normalized_name = self._find_normalized_tool_name(self._tool_name)
    tool_id = f"tooluse_{self._tool_name}_{random.randint(100000000, 999999999)}"
    tool_use: ToolUse = {"toolUseId": tool_id, "name": normalized_name, "input": kwargs.copy()}
    return tool_use, [], kwargs

[github-actions]

Assessment: Request Changes

This PR addresses a real gap in the developer experience (issue #1436) and the architectural approach of wrapping ToolExecutor._stream() is sound. However, there are several issues that should be addressed before merging.

Review Categories

• Critical — stream() doesn't actually stream: The sync stream() method buffers all events before yielding, making it functionally identical to __call__() but marketed as streaming. This violates the "obvious path is the happy path" tenet — either implement true sync streaming, remove it, or rename to make the behavior clear.
• API Review Process: This PR adds new public API methods (stream / stream_async) but lacks the needs-api-review label. Key design questions remain around parameter parity with __call__(), the no-recording-by-default decision, and TypedEvent not being publicly exported.
• Documentation PR: This adds new public API surface (streaming methods on tool calls). Per the documentation criteria, a documentation PR is required for new features that add public API surface. The PR claims "No documentation PR needed" but this should be reconsidered.
• Behavioral asymmetry: stream_async() skips concurrency guards and conversation management that __call__() performs — this should be an intentional, documented design decision rather than an omission.
• Code duplication: Tool setup logic is duplicated between __call__() and stream_async() and should be extracted.

The refactoring of _ToolCaller to return _ToolExecutor instances is a clean approach that preserves backward compatibility.

[github-actions]

Issue: The __call__ method applies conversation_manager.apply_management() after execution (line 134), but stream_async() does not. While this may be intentional since streaming doesn't record to history, if a user mixes streaming and non-streaming calls, the conversation management behavior could become inconsistent.

Suggestion: Document this behavioral difference explicitly, or keep the two paths symmetric.