Class ToolProcessor

src/mcp_cli/chat/tool_processor.py:55–1679

Handle execution of tool calls returned by the LLM. Delegates parallel execution to ToolManager.stream_execute_tools(), handling only CLI-specific concerns: UI, conversation history, confirmation. Uses ToolProcessorContext protocol for type-safe context access.

class ToolProcessor:
    """
    Handle execution of tool calls returned by the LLM.

    Delegates parallel execution to ToolManager.stream_execute_tools(),
    handling only CLI-specific concerns: UI, conversation history, confirmation.

    Uses ToolProcessorContext protocol for type-safe context access.
    """

    def __init__(
        self,
        context: ToolProcessorContext,
        ui_manager: UIManagerProtocol,
        *,
        max_concurrency: int = 4,
    ) -> None:
        self.context = context
        self.ui_manager = ui_manager
        self.max_concurrency = max_concurrency

        # Tool manager for execution - access via protocol attribute
        self.tool_manager: ToolManager | None = context.tool_manager

        # Track transport failures for recovery detection
        self._transport_failures = 0
        self._consecutive_transport_failures = 0

        # Track state for callbacks
        self._call_metadata: dict[str, ToolCallMetadata] = {}
        self._cancelled = False

        # Track which tool_call_ids have received results (for orphan detection)
        self._result_ids_added: set[str] = set()

        # Track page_fault calls within a conversation to prevent re-fault loops
        self._faulted_page_ids: set[str] = set()

        # Give the context a back-pointer for Ctrl-C cancellation
        # Note: This is the one place we set an attribute on context
        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.")