Method _execute_step

src/mcp_cli/planning/executor.py:339–388

Execute a single plan step using the LLM for tool call generation. When a model_manager is available, the LLM generates the tool call from the step description and tool schemas, then the tool is executed. If the tool fails, the error is fed back to the LLM for retry.

(
        self,
        step: dict[str, Any],
        var_context: dict[str, Any],
    )

            return self._tool_catalog

    async def _execute_step(
        self,
        step: dict[str, Any],
        var_context: dict[str, Any],
    ) -> StepResult:
        """Execute a single plan step using the LLM for tool call generation.

        When a model_manager is available, the LLM generates the tool call
        from the step description and tool schemas, then the tool is executed.
        If the tool fails, the error is fed back to the LLM for retry.

        Falls back to static arg execution when no model_manager is provided.
        """
        step_index = step.get("index", "?")
        step_title = step.get("title", "Untitled")
        tool_calls = step.get("tool_calls", [])
        hint_tool = tool_calls[0]["name"] if tool_calls else step.get("tool", "none")
        hint_args = (
            tool_calls[0].get("args", {}) if tool_calls else step.get("args", {})
        )

        if self._on_step_start:
            self._on_step_start(step_index, step_title, hint_tool)

        start_time = time.perf_counter()

        # LLM-driven execution (agentic loop with retry)
        if self._model_manager:
            logger.info(
                "Step %s: using agentic LLM execution (model_manager=%s)",
                step_index,
                type(self._model_manager).__name__,
            )
            step_result = await self._execute_step_with_llm(
                step, var_context, step_index, step_title, hint_tool, hint_args
            )
        else:
            logger.info(
                "Step %s: using static arg execution (no model_manager)", step_index
            )
            step_result = await self._execute_step_static(
                step, var_context, step_index, step_title, hint_tool, hint_args
            )

        step_result.duration = time.perf_counter() - start_time

        if self._on_step_complete:
            self._on_step_complete(step_result)

        return step_result

    async def _execute_step_static(
        self,