Function execute_single

src/mcp_cli/tools/execution.py:66–111

Execute a single tool call with semaphore control.

(call: CTPToolCall)

    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)

            # Convert ToolCallResult to CTPToolResult
            ctp_result = CTPToolResult(
                id=call.id,
                tool=call.tool,
                result=tool_result.result if tool_result.success else None,
                error=tool_result.error if not tool_result.success else None,
                start_time=start_time,
                end_time=end_time,
                machine=platform.node(),
                pid=os.getpid(),
            )

            # Invoke result callback
            if on_tool_result:
                try:
                    await on_tool_result(ctp_result)
                except Exception as e:
                    logger.warning(
                        f"on_tool_result callback failed for {call.tool}: {e}"
                    )

            return ctp_result

    # Create all tasks and execute in parallel
    tasks = [asyncio.create_task(execute_single(call)) for call in calls]