Method process_tool_calls

src/mcp_cli/chat/tool_processor.py:97–437

Execute tool_calls in parallel using ToolManager.stream_execute_tools(). Args: tool_calls: List of tool call objects from the LLM name_mapping: Mapping from LLM tool names to actual tool names reasoning_content: Optional reasoning content from th

(
        self,
        tool_calls: list[Any],
        name_mapping: dict[str, str] | None = None,
        reasoning_content: str | None = None,
    )

        context.tool_processor = self

    async def process_tool_calls(
        self,
        tool_calls: list[Any],
        name_mapping: dict[str, str] | None = None,
        reasoning_content: str | None = None,
    ) -> None:
        """
        Execute tool_calls in parallel using ToolManager.stream_execute_tools().

        Args:
            tool_calls: List of tool call objects from the LLM
            name_mapping: Mapping from LLM tool names to actual tool names
            reasoning_content: Optional reasoning content from the LLM
        """
        if not tool_calls:
            logger.warning("Empty tool_calls list received.")
            return

        if name_mapping is None:
            name_mapping = {}

        logger.info(
            f"Processing {len(tool_calls)} tool calls with {len(name_mapping)} name mappings"
        )

        # Reset state
        self._call_metadata.clear()
        self._cancelled = False
        self._result_ids_added = set()
        self._faulted_page_ids.clear()

        # Add assistant message with all tool calls BEFORE executing
        self._add_assistant_message_with_tool_calls(tool_calls, reasoning_content)

        # Convert LLM tool calls to CTP format and check confirmations
        ctp_calls: list[CTPToolCall] = []

        try:
            for idx, call in enumerate(tool_calls):
                if getattr(self.ui_manager, "interrupt_requested", False):
                    self._cancelled = True
                    break

                # Extract tool call details
                llm_tool_name, raw_arguments, call_id = self._extract_tool_call_info(
                    call, idx
                )

                # Map to execution name
                execution_tool_name = name_mapping.get(llm_tool_name, llm_tool_name)

                # Get display name - special handling for dynamic tool call_tool
                display_name = execution_tool_name
                display_arguments = raw_arguments

                # For dynamic tools, extract the actual tool name from call_tool
                if execution_tool_name == DYNAMIC_TOOL_PROXY_NAME:
                    # Parse arguments to get the real tool name