[FEATURE] Concurrent Direct Tool Calls

[namontalbano]

Problem Statement

Summary

Add support for executing multiple direct tool calls concurrently via agent.tool.*, allowing callers to fan out independent tool executions in parallel rather than sequentially.

---

Background

The Strands SDK exposes a convenient direct tool call interface via agent.tool.<tool_name>(**kwargs). Internally, _ToolCaller.__getattr__ returns a synchronous caller function that wraps an async executor and calls run_async(acall) (_caller.py:124).

When record_direct_tool_call=True, the caller acquires _invocation_lock before executing and releases it in the finally block. This serializes all recorded direct tool calls by design — only one can hold the lock at a time.

acquired_lock = (
    should_lock
    and isinstance(self._agent, Agent)
    and self._agent._invocation_lock.acquire_lock(blocking=False)
)
if should_lock and not acquired_lock:
    raise ConcurrencyException(
        "Direct tool call cannot be made while the agent is in the middle of an invocation. "
        "Set record_direct_tool_call=False to allow direct tool calls during agent invocation."
    )

---

Problem

There is no built-in way to call multiple tools concurrently. Callers who need to fan out independent tool calls — e.g., fetching data from several sources simultaneously — must execute them one at a time:

result_a = agent.tool.fetch_data(source="a")
result_b = agent.tool.fetch_data(source="b")
result_c = agent.tool.fetch_data(source="c")
# Total latency = latency_a + latency_b + latency_c

This is especially costly when tools involve I/O (HTTP calls, database queries, knowledge base lookups). Our project frequently needs to query multiple knowledge bases or run multiple enrichment tools before composing a final response, and sequential execution multiplies end-to-end latency.

---

Implementation sketch:

• Refactor the inner acall() coroutine in __getattr__ into a standalone async method on _ToolCaller (e.g., _acall_tool).
• In concurrent, acquire the lock once for the entire batch (if record_direct_tool_call=True), then asyncio.gather all the inner coroutines.
• After all results are collected, record them as a single batched message sequence — similar to how the LLM handles parallel tool use in a single assistant turn.

---

Proposed Solution

Option A: agent.tool.concurrent(*calls) helper (preferred)

Add a concurrent method to _ToolCaller that accepts an iterable of (tool_name, kwargs) pairs (or pre-built callables), executes them concurrently under a shared async gather, and returns results in the same order:

results = await agent.tool.concurrent(
    ("fetch_data", {"source": "a"}),
    ("fetch_data", {"source": "b"}),
    ("fetch_data", {"source": "c"}),
)
# Total latency ≈ max(latency_a, latency_b, latency_c)

Or using a more fluent API:

results = agent.tool.concurrent(
    agent.tool.fetch_data.build(source="a"),
    agent.tool.fetch_data.build(source="b"),
    agent.tool.fetch_data.build(source="c"),
)

Option B: Async-native agent.tool.<name> (alternative)

Make the caller function returned by __getattr__ natively async and let callers use asyncio.gather directly:

results = await asyncio.gather(
    agent.tool.fetch_data(source="a"),
    agent.tool.fetch_data(source="b"),
)

Use Case

# Current: sequential KB lookups add up to 3–4s per query
kb_result = agent.tool.knowledge_base_lookup(query=user_query, kb_id="general")
faq_result = agent.tool.knowledge_base_lookup(query=user_query, kb_id="faq")
product_result = agent.tool.knowledge_base_lookup(query=user_query, kb_id="products")

# Desired: concurrent, ~1s total
kb_result, faq_result, product_result = agent.tool.concurrent(
    ("knowledge_base_lookup", {"query": user_query, "kb_id": "general"}),
    ("knowledge_base_lookup", {"query": user_query, "kb_id": "faq"}),
    ("knowledge_base_lookup", {"query": user_query, "kb_id": "products"}),
)

Additional Context

References

• Current implementation: strands/tools/_caller.py
• ConcurrencyException raised today when lock cannot be acquired: strands/types/exceptions.py