Execute a plan tool (plan_create, plan_execute, plan_create_and_execute). Plan tools are internal operations that bypass the MCP ToolManager and all guard checks. They use PlanningContext to generate and execute multi-step plans.
(
self,
tool_name: str,
arguments: dict,
llm_tool_name: str,
call_id: str,
)
| 723 | self._add_tool_result_to_history(llm_tool_name, call_id, result_text) |
| 724 | |
| 725 | async def _handle_plan_tool( |
| 726 | self, |
| 727 | tool_name: str, |
| 728 | arguments: dict, |
| 729 | llm_tool_name: str, |
| 730 | call_id: str, |
| 731 | ) -> None: |
| 732 | """Execute a plan tool (plan_create, plan_execute, plan_create_and_execute). |
| 733 | |
| 734 | Plan tools are internal operations that bypass the MCP ToolManager |
| 735 | and all guard checks. They use PlanningContext to generate and |
| 736 | execute multi-step plans. |
| 737 | """ |
| 738 | if not getattr(self.context, "_enable_plan_tools", False): |
| 739 | self._add_tool_result_to_history( |
| 740 | llm_tool_name, call_id, "Plan tools are not enabled." |
| 741 | ) |
| 742 | return |
| 743 | |
| 744 | logger.info("Plan tool %s called with args: %s", tool_name, arguments) |
| 745 | |
| 746 | from mcp_cli.planning.context import PlanningContext |
| 747 | from mcp_cli.planning.tools import handle_plan_tool |
| 748 | |
| 749 | # Lazy-create PlanningContext (cached on context object) |
| 750 | planning_context = getattr(self.context, "_planning_context", None) |
| 751 | if planning_context is None: |
| 752 | planning_context = PlanningContext(self.context.tool_manager) |
| 753 | self.context._planning_context = planning_context |
| 754 | |
| 755 | # Get model_manager for LLM-driven step execution |
| 756 | model_manager = getattr(self.context, "model_manager", None) |
| 757 | |
| 758 | # Broadcast plan start to dashboard |
| 759 | bridge = getattr(self.context, "dashboard_bridge", None) |
| 760 | plan_title = arguments.get("goal", "Plan") |
| 761 | if bridge is not None: |
| 762 | try: |
| 763 | await bridge.on_plan_update( |
| 764 | plan_id=call_id, |
| 765 | title=plan_title, |
| 766 | steps=[], |
| 767 | status="running", |
| 768 | ) |
| 769 | except Exception as _e: |
| 770 | logger.debug("Dashboard plan start update error: %s", _e) |
| 771 | |
| 772 | # Pass the UI manager so handle_plan_tool can show step-by-step progress |
| 773 | result_text = await handle_plan_tool( |
| 774 | tool_name, |
| 775 | arguments, |
| 776 | planning_context, |
| 777 | model_manager, |
| 778 | ui_manager=self.ui_manager, |
| 779 | ) |
| 780 | |
| 781 | # Re-fetch bridge in case it changed during plan execution |
| 782 | bridge = getattr(self.context, "dashboard_bridge", None) |
no test coverage detected