Function stream_execute_tools

src/mcp_cli/tools/execution.py:146–263

Execute multiple tool calls in parallel, yielding results as they complete. This is the streaming version of execute_tools_parallel - results are yielded immediately when each tool completes, without waiting for all. Args: manager: ToolManager instance to execute tools

(
    manager: ToolManager,
    calls: list[CTPToolCall],
    timeout: float | None = None,
    on_tool_start: Callable[[CTPToolCall], Awaitable[None]] | None = None,
    max_concurrency: int = 4,
    batch_timeout: float | None = None,
)

async def stream_execute_tools(
    manager: ToolManager,
    calls: list[CTPToolCall],
    timeout: float | None = None,
    on_tool_start: Callable[[CTPToolCall], Awaitable[None]] | None = None,
    max_concurrency: int = 4,
    batch_timeout: float | None = None,
) -> AsyncIterator[CTPToolResult]:
    """
    Execute multiple tool calls in parallel, yielding results as they complete.

    This is the streaming version of execute_tools_parallel - results are
    yielded immediately when each tool completes, without waiting for all.

    Args:
        manager: ToolManager instance to execute tools
        calls: List of CTPToolCall objects to execute
        timeout: Timeout per tool execution (uses default if not specified)
        on_tool_start: Async callback invoked when each tool starts
        max_concurrency: Maximum concurrent executions (default: 4)
        batch_timeout: Global timeout for entire batch (auto-computed if None)

    Yields:
        CTPToolResult objects as each tool completes (in completion order)
    """
    if not calls:
        return

    effective_timeout = timeout or manager.tool_timeout
    sem = asyncio.Semaphore(max_concurrency)
    queue: asyncio.Queue[CTPToolResult] = asyncio.Queue()

    async def execute_single(call: CTPToolCall) -> None:
        """Execute a single tool call and put result in queue."""
        start_time = datetime.now(UTC)

        async with sem:
            # Invoke start callback
            if on_tool_start:
                try:
                    await on_tool_start(call)
                except Exception as e:
                    logger.warning(
                        f"on_tool_start callback failed for {call.tool}: {e}"
                    )

            # Execute the tool
            tool_result = await manager.execute_tool(
                call.tool,
                call.arguments,
                namespace=call.namespace if call.namespace != "default" else None,
                timeout=effective_timeout,
            )

            end_time = datetime.now(UTC)

            # Convert ToolCallResult to CTPToolResult
            ctp_result = CTPToolResult(