Function execute_tools_parallel

src/mcp_cli/tools/execution.py:32–143

Execute multiple tool calls in parallel with optional callbacks. Uses chuk-tool-processor's ToolCall/ToolResult models for consistency. Results are returned in completion order (faster tools return first). Args: manager: ToolManager instance to execute tools calls:

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

async def execute_tools_parallel(
    manager: ToolManager,
    calls: list[CTPToolCall],
    timeout: float | None = None,
    on_tool_start: Callable[[CTPToolCall], Awaitable[None]] | None = None,
    on_tool_result: Callable[[CTPToolResult], Awaitable[None]] | None = None,
    max_concurrency: int = 4,
    batch_timeout: float | None = None,
) -> list[CTPToolResult]:
    """
    Execute multiple tool calls in parallel with optional callbacks.

    Uses chuk-tool-processor's ToolCall/ToolResult models for consistency.
    Results are returned in completion order (faster tools return first).

    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
        on_tool_result: Async callback invoked when each tool completes
        max_concurrency: Maximum concurrent executions (default: 4)
        batch_timeout: Global timeout for entire batch (auto-computed if None)

    Returns:
        List of CTPToolResult objects in completion order
    """
    if not calls:
        return []

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

    async def execute_single(call: CTPToolCall) -> CTPToolResult:
        """Execute a single tool call with semaphore control."""
        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)