Release v2.3.5

- Add planning module with PlanningAgent, Plan, PlanStep, TodoList, TodoItem
- Add PlanStorage for persisting plans to .praison/plans/
- Add ApprovalCallback for plan approval flow
- Add plan_mode to Agent for read-only tool filtering
- Add planning, planning_llm, auto_approve_plan params to PraisonAIAgents
- Add 71 comprehensive tests for planning mode
- Update READMEs with Planning Mode documentation
- Bump version to 0.0.181

README.md

@@ -60,6 +60,7 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
 - 💾 **Prompt Caching** (Reduce costs & latency - OpenAI, Anthropic, Bedrock, Deepseek)
 - 🧠 **Claude Memory Tool** (Persistent cross-conversation memory - Anthropic Beta)
 - 💾 **File-Based Memory** (Zero-dependency persistent memory for all agents)
+- 📋 **Planning Mode** (Plan before execution like Cursor, Windsurf, Claude Code)
 
 ## Using Python Code
 

docker/Dockerfile

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     flask \
-    "praisonai>=2.3.4" \
+    "praisonai>=2.3.5" \
     "praisonai[api]" \
     gunicorn \
     markdown

docker/Dockerfile.chat

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=2.3.4" \
+    "praisonai>=2.3.5" \
     "praisonai[chat]" \
     "embedchain[github,youtube]"
 

docker/Dockerfile.dev

@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=2.3.4" \
+    "praisonai>=2.3.5" \
     "praisonai[ui]" \
     "praisonai[chat]" \
     "praisonai[realtime]" \

docker/Dockerfile.ui

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=2.3.4" \
+    "praisonai>=2.3.5" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

src/praisonai-agents/README.md

@@ -84,6 +84,38 @@ agents = PraisonAIAgents(agents=[researcher, writer], tasks=[research_task, writ
 result = agents.start()
 ```
 
+### Planning Mode 🆕
+
+Plan before execution, like Cursor, Windsurf, and Claude Code:
+
+```python
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+researcher = Agent(name="Researcher", role="Research Analyst")
+writer = Agent(name="Writer", role="Content Writer")
+
+research_task = Task(description="Research AI benefits", agent=researcher)
+write_task = Task(description="Write summary", agent=writer)
+
+# Enable planning mode
+agents = PraisonAIAgents(
+    agents=[researcher, writer],
+    tasks=[research_task, write_task],
+    planning=True,              # Enable planning
+    planning_llm="gpt-4o-mini", # Fast LLM for planning
+    auto_approve_plan=True      # Auto-approve plans
+)
+
+result = agents.start()
+```
+
+**Features:**
+- **Plan Creation**: LLM creates step-by-step implementation plans
+- **Todo Lists**: Auto-generated from plans with progress tracking
+- **Read-Only Mode**: `plan_mode=True` restricts agents to safe tools
+- **Approval Flow**: Review and approve plans before execution
+- **Persistence**: Plans saved to `.praison/plans/` as markdown
+
 ---
 
 ## Agent Types

src/praisonai-agents/planning_example.py

@@ -0,0 +1,34 @@
+"""Simple Planning Mode Example: Plan first, then execute todo items."""
+
+from praisonaiagents import Agent
+from praisonaiagents.planning import PlanningAgent, TodoList
+
+# Step 1: Create agents
+researcher = Agent(name="Researcher", role="Research Analyst", verbose=True)
+writer = Agent(name="Writer", role="Content Writer", verbose=True)
+
+# Step 2: Create a plan using PlanningAgent
+planner = PlanningAgent(llm="gpt-4o-mini", verbose=1)
+plan = planner.create_plan_sync(
+    request="Write a short article about the top 3 benefits of meditation",
+    agents=[researcher, writer],
+    context="Keep it concise"
+)
+
+# Step 3: Create todo list from the plan
+todo = TodoList.from_plan(plan)
+
+# Step 4: Execute each todo item one by one
+for item in todo.items:
+    todo.start(item.id)
+    
+    # Pick the right agent based on the item's assigned agent
+    if item.agent == "Researcher":
+        agent = researcher
+    else:
+        agent = writer
+    
+    # Execute the task
+    result = agent.chat(item.description)
+    
+    todo.complete(item.id)

src/praisonai-agents/planning_simple.py

@@ -0,0 +1,21 @@
+"""Simplest Planning Mode Example - Just enable planning=True"""
+
+from praisonaiagents import Agent, Task, PraisonAIAgents
+
+# Create agents
+researcher = Agent(name="Researcher", role="Research Analyst")
+writer = Agent(name="Writer", role="Content Writer")
+
+# Create tasks
+task1 = Task(description="Research the benefits of meditation", agent=researcher)
+task2 = Task(description="Write a short article about meditation benefits", agent=writer)
+
+# Run with planning enabled - that's it!
+agents = PraisonAIAgents(
+    agents=[researcher, writer],
+    tasks=[task1, task2],
+    planning=True,
+    auto_approve_plan=True  # Auto-approve the generated plan
+)
+
+agents.start()

src/praisonai-agents/praisonaiagents/__init__.py

@@ -55,6 +55,32 @@ except ImportError:
 from .session import Session
 from .memory.memory import Memory
 from .guardrails import GuardrailResult, LLMGuardrail
+
+# Planning mode support
+try:
+    from .planning import (
+        Plan,
+        PlanStep,
+        TodoList,
+        TodoItem,
+        PlanStorage,
+        PlanningAgent,
+        ApprovalCallback,
+        READ_ONLY_TOOLS,
+        RESTRICTED_TOOLS,
+    )
+    _planning_available = True
+except ImportError:
+    _planning_available = False
+    Plan = None
+    PlanStep = None
+    TodoList = None
+    TodoItem = None
+    PlanStorage = None
+    PlanningAgent = None
+    ApprovalCallback = None
+    READ_ONLY_TOOLS = []
+    RESTRICTED_TOOLS = []
 from .agent.handoff import Handoff, handoff, handoff_filters, RECOMMENDED_PROMPT_PREFIX, prompt_with_handoff_instructions
 # Flow display
 try:

src/praisonai-agents/praisonaiagents/agent/agent.py

@@ -238,7 +238,8 @@ class Agent:
         web_search: Optional[Union[bool, Dict[str, Any]]] = None,
         web_fetch: Optional[Union[bool, Dict[str, Any]]] = None,
         prompt_caching: Optional[bool] = None,
-        claude_memory: Optional[Union[bool, Any]] = None
+        claude_memory: Optional[Union[bool, Any]] = None,
+        plan_mode: bool = False
     ):
         """Initialize an Agent instance.
 
@@ -524,6 +525,7 @@ Your Goal: {self.goal}
         # Store user_id
         self.user_id = user_id or "praison"
         self.reasoning_steps = reasoning_steps
+        self.plan_mode = plan_mode  # Read-only mode for planning
         self.web_search = web_search
         self.web_fetch = web_fetch
         self.prompt_caching = prompt_caching
@@ -610,6 +612,37 @@ Your Goal: {self.goal}
             self._agent_id = str(uuid.uuid4())
         return self._agent_id
     
+    def get_available_tools(self) -> List[Any]:
+        """
+        Get tools available to this agent, filtered by plan_mode if enabled.
+        
+        In plan_mode, only read-only tools are available to prevent
+        modifications during the planning phase.
+        
+        Returns:
+            List of available tools
+        """
+        if not self.plan_mode:
+            return self.tools
+            
+        # Filter to read-only tools only
+        from ..planning import READ_ONLY_TOOLS, RESTRICTED_TOOLS
+        
+        filtered_tools = []
+        for tool in self.tools:
+            tool_name = getattr(tool, '__name__', str(tool)).lower()
+            
+            # Check if tool is in restricted list
+            is_restricted = any(
+                restricted.lower() in tool_name 
+                for restricted in RESTRICTED_TOOLS
+            )
+            
+            if not is_restricted:
+                filtered_tools.append(tool)
+                
+        return filtered_tools
+    
     def _model_supports_web_search(self) -> bool:
         """
         Check if the agent's model supports native web search via LiteLLM.

src/praisonai-agents/praisonaiagents/agents/agents.py

@@ -127,7 +127,7 @@ def process_task_context(context_item, verbose=0, user_id=None):
         return str(context_item)  # Fallback for unknown types
 
 class PraisonAIAgents:
-    def __init__(self, agents, tasks=None, verbose=0, completion_checker=None, max_retries=5, process="sequential", manager_llm=None, memory=False, memory_config=None, embedder=None, user_id=None, max_iter=10, stream=True, name: Optional[str] = None):
+    def __init__(self, agents, tasks=None, verbose=0, completion_checker=None, max_retries=5, process="sequential", manager_llm=None, memory=False, memory_config=None, embedder=None, user_id=None, max_iter=10, stream=True, name: Optional[str] = None, planning: bool = False, planning_llm: Optional[str] = None, auto_approve_plan: bool = False):
         # Add check at the start if memory is requested
         if memory:
             try:
@@ -213,6 +213,14 @@ class PraisonAIAgents:
         
         self._state = {}  # Add state storage at PraisonAIAgents level
         
+        # Initialize planning mode
+        self.planning = planning
+        self.planning_llm = planning_llm or "gpt-4o-mini"
+        self.auto_approve_plan = auto_approve_plan
+        self._current_plan = None
+        self._todo_list = None
+        self._planning_agent = None
+        
         # Initialize memory system
         self.shared_memory = None
         if memory:
@@ -929,9 +937,13 @@ Context:
                         task.context = []
                     # Add content to context
                     task.context.append(content)
-                
-        # Run tasks as before
-        self.run_all_tasks()
+        
+        # Planning Mode: Create plan and todo list before execution
+        if self.planning:
+            self._run_with_planning()
+        else:
+            # Run tasks as before
+            self.run_all_tasks()
         
         # Auto-display token metrics if any agent has metrics=True
         metrics_enabled = any(getattr(agent, 'metrics', False) for agent in self.agents)
@@ -1532,3 +1544,352 @@ Context:
         else:
             display_error(f"Invalid protocol: {protocol}. Choose 'http' or 'mcp'.")
             return None
+
+    # =========================================================================
+    # Planning Mode Properties and Methods
+    # =========================================================================
+    
+    @property
+    def current_plan(self):
+        """Get the current plan."""
+        return self._current_plan
+    
+    @property
+    def todo_list(self):
+        """Get the current todo list."""
+        return self._todo_list
+    
+    def _get_planning_agent(self):
+        """Lazy load PlanningAgent."""
+        if self._planning_agent is None and self.planning:
+            from ..planning import PlanningAgent
+            self._planning_agent = PlanningAgent(
+                llm=self.planning_llm,
+                read_only=True,
+                verbose=self.verbose
+            )
+        return self._planning_agent
+    
+    async def _create_plan(self, request: str = None, context: str = None):
+        """
+        Create an implementation plan.
+        
+        Args:
+            request: The request/goal to plan for
+            context: Optional additional context
+            
+        Returns:
+            Plan instance
+        """
+        planner = self._get_planning_agent()
+        if planner is None:
+            return None
+            
+        # Use first task description as request if not provided
+        if request is None and self.tasks:
+            first_task = list(self.tasks.values())[0]
+            request = first_task.description
+            
+        plan = await planner.create_plan(
+            request=request,
+            agents=self.agents,
+            tasks=list(self.tasks.values()),
+            context=context
+        )
+        
+        self._current_plan = plan
+        
+        # Create todo list from plan
+        from ..planning import TodoList
+        self._todo_list = TodoList.from_plan(plan)
+        
+        return plan
+    
+    def _create_plan_sync(self, request: str = None, context: str = None):
+        """
+        Synchronous version of _create_plan.
+        
+        Args:
+            request: The request/goal to plan for
+            context: Optional additional context
+            
+        Returns:
+            Plan instance
+        """
+        planner = self._get_planning_agent()
+        if planner is None:
+            return None
+            
+        # Use first task description as request if not provided
+        if request is None and self.tasks:
+            first_task = list(self.tasks.values())[0]
+            request = first_task.description
+            
+        plan = planner.create_plan_sync(
+            request=request,
+            agents=self.agents,
+            tasks=list(self.tasks.values()),
+            context=context
+        )
+        
+        self._current_plan = plan
+        
+        # Create todo list from plan
+        from ..planning import TodoList
+        self._todo_list = TodoList.from_plan(plan)
+        
+        return plan
+    
+    async def _request_approval(self, plan):
+        """
+        Request approval for a plan.
+        
+        Args:
+            plan: Plan to approve
+            
+        Returns:
+            True if approved, False if rejected
+        """
+        if self.auto_approve_plan:
+            plan.approve()
+            return True
+            
+        from ..planning import ApprovalCallback
+        callback = ApprovalCallback(auto_approve=self.auto_approve_plan)
+        return await callback.async_call(plan)
+    
+    def _request_approval_sync(self, plan):
+        """
+        Synchronous version of _request_approval.
+        
+        Args:
+            plan: Plan to approve
+            
+        Returns:
+            True if approved, False if rejected
+        """
+        if self.auto_approve_plan:
+            plan.approve()
+            return True
+            
+        from ..planning import ApprovalCallback
+        callback = ApprovalCallback(auto_approve=self.auto_approve_plan)
+        return callback(plan)
+    
+    def get_plan_markdown(self) -> str:
+        """
+        Get the current plan as markdown.
+        
+        Returns:
+            Markdown string or empty string if no plan
+        """
+        if self._current_plan:
+            return self._current_plan.to_markdown()
+        return ""
+    
+    def get_todo_markdown(self) -> str:
+        """
+        Get the current todo list as markdown.
+        
+        Returns:
+            Markdown string or empty string if no todo list
+        """
+        if self._todo_list:
+            return self._todo_list.to_markdown()
+        return ""
+    
+    def update_plan_step_status(self, step_id: str, status: str) -> bool:
+        """
+        Update the status of a plan step.
+        
+        Args:
+            step_id: ID of the step to update
+            status: New status
+            
+        Returns:
+            True if updated, False if not found
+        """
+        if self._current_plan:
+            result = self._current_plan.update_step_status(step_id, status)
+            # Sync todo list
+            if self._todo_list:
+                self._todo_list.sync_with_plan(self._current_plan)
+            return result
+        return False
+    
+    def _run_with_planning(self):
+        """
+        Run tasks with planning mode enabled.
+        
+        This method:
+        1. Creates a plan using PlanningAgent
+        2. Generates a todo list from the plan
+        3. Creates proper Task objects from plan steps
+        4. Executes each task using the full Task execution system
+           (with memory, callbacks, guardrails, structured output, etc.)
+        5. Tracks progress as items are completed
+        """
+        from rich.console import Console
+        from rich.panel import Panel
+        from rich.markdown import Markdown
+        from ..task import Task
+        
+        console = Console()
+        
+        # Step 1: Create the plan
+        console.print("\n[bold blue]📋 PLANNING PHASE[/bold blue]")
+        console.print("[dim]Creating implementation plan...[/dim]\n")
+        
+        # Build request from tasks
+        task_descriptions = [task.description for task in self.tasks.values()]
+        request = " AND ".join(task_descriptions)
+        
+        plan = self._create_plan_sync(request=request)
+        
+        if not plan:
+            console.print("[yellow]⚠️ Planning failed, falling back to normal execution[/yellow]")
+            self.run_all_tasks()
+            return
+        
+        # Display the plan
+        console.print(Panel(
+            Markdown(plan.to_markdown()),
+            title="[bold green]Generated Plan[/bold green]",
+            border_style="green"
+        ))
+        
+        # Step 2: Request approval if not auto-approve
+        if not self.auto_approve_plan:
+            approved = self._request_approval_sync(plan)
+            if not approved:
+                console.print("[red]❌ Plan rejected. Aborting execution.[/red]")
+                return
+        else:
+            plan.approve()
+            console.print("[green]✅ Plan auto-approved[/green]")
+        
+        # Step 3: Create todo list for tracking
+        from ..planning import TodoList
+        self._todo_list = TodoList.from_plan(plan)
+        
+        console.print("\n[bold blue]📝 TODO LIST[/bold blue]")
+        console.print(Panel(
+            Markdown(self._todo_list.to_markdown()),
+            title="[bold cyan]Tasks to Complete[/bold cyan]",
+            border_style="cyan"
+        ))
+        
+        # Step 4: Create proper Task objects from plan steps
+        console.print("\n[bold blue]🚀 EXECUTION PHASE[/bold blue]\n")
+        
+        # Map agent names to agent instances
+        agent_map = {agent.name: agent for agent in self.agents}
+        
+        # Store original tasks and create new tasks from plan
+        original_tasks = self.tasks.copy()
+        self.tasks = {}
+        self.task_id_counter = 0
+        
+        # Create Task objects from plan steps
+        plan_tasks = []
+        step_to_task = {}  # Map step_id to task for context chaining
+        
+        for i, step in enumerate(plan.steps):
+            # Get the appropriate agent
+            agent = agent_map.get(step.agent, self.agents[0] if self.agents else None)
+            
+            if not agent:
+                console.print(f"[yellow]⚠️ No agent found for '{step.agent}', using first available[/yellow]")
+                agent = self.agents[0] if self.agents else None
+            
+            if not agent:
+                console.print(f"[red]❌ No agents available for step: {step.description}[/red]")
+                continue
+            
+            # Build context from dependencies (previous task results)
+            context = []
+            for dep_id in step.dependencies:
+                # Convert step_X format to actual step index
+                if dep_id.startswith("step_"):
+                    try:
+                        dep_index = int(dep_id.split("_")[1])
+                        if dep_index < len(plan_tasks):
+                            context.append(plan_tasks[dep_index])
+                    except (ValueError, IndexError):
+                        pass
+                elif dep_id in step_to_task:
+                    context.append(step_to_task[dep_id])
+            
+            # Find matching original task for additional config (memory, callbacks, etc.)
+            original_task = None
+            for orig_task in original_tasks.values():
+                if orig_task.agent and orig_task.agent.name == agent.name:
+                    original_task = orig_task
+                    break
+            
+            # Create Task with full features from original task if available
+            task = Task(
+                description=step.description,
+                expected_output=f"Complete: {step.description}",
+                agent=agent,
+                name=f"Plan Step {i + 1}",
+                tools=agent.tools if agent.tools else [],
+                context=context if context else None,
+                # Inherit from original task if available
+                memory=original_task.memory if original_task else None,
+                callback=original_task.callback if original_task else None,
+                guardrail=original_task.guardrail if original_task else None,
+                max_retries=original_task.max_retries if original_task else 3,
+                output_json=original_task.output_json if original_task else None,
+                output_pydantic=original_task.output_pydantic if original_task else None,
+                config=original_task.config if original_task else {}
+            )
+            
+            # Add task to our task list
+            task_id = self.add_task(task)
+            plan_tasks.append(task)
+            step_to_task[step.id] = task
+        
+        # Step 5: Execute tasks using the proper Task execution system
+        for i, (task_id, task) in enumerate(self.tasks.items()):
+            # Update todo list progress
+            if i < len(self._todo_list.items):
+                item = self._todo_list.items[i]
+                
+                # Display progress bar
+                progress = self._todo_list.progress
+                bar_length = 30
+                filled = int(bar_length * progress)
+                bar = "█" * filled + "░" * (bar_length - filled)
+                console.print(f"[dim]Progress: [{bar}] {progress * 100:.0f}%[/dim]")
+                
+                console.print(f"\n[bold]📌 Step {i + 1}/{len(self.tasks)}:[/bold] {task.description[:60]}...")
+                console.print(f"[dim]   Agent: {task.agent.name if task.agent else 'Unknown'}[/dim]")
+                
+                # Mark as in progress
+                self._todo_list.start(item.id)
+            
+            # Execute using the full Task execution system
+            # This includes: memory, callbacks, guardrails, structured output, retry logic
+            try:
+                self.run_task(task_id)
+                
+                if task.status == "completed":
+                    if i < len(self._todo_list.items):
+                        self._todo_list.complete(self._todo_list.items[i].id)
+                    console.print("[green]   ✅ Completed[/green]")
+                else:
+                    console.print(f"[yellow]   ⚠️ Task status: {task.status}[/yellow]")
+            except Exception as e:
+                console.print(f"[red]   ❌ Error: {e}[/red]")
+                logger.error(f"Error executing plan task {task_id}: {e}")
+        
+        # Final progress
+        completed_count = len([t for t in self.tasks.values() if t.status == "completed"])
+        console.print(f"\n[bold green]🎉 EXECUTION COMPLETE[/bold green]")
+        console.print(f"[dim]Progress: [{'█' * 30}] 100%[/dim]")
+        console.print(f"[green]Completed {completed_count}/{len(self.tasks)} tasks![/green]\n")
+        
+        # Restore original tasks reference for result retrieval
+        self._plan_tasks = self.tasks.copy()
+        # Keep plan tasks for results but note original tasks are preserved in _plan_tasks

src/praisonai-agents/praisonaiagents/planning/__init__.py

@@ -0,0 +1,86 @@
+"""
+Planning Module for PraisonAI Agents.
+
+Provides Planning Mode functionality similar to:
+- Cursor Plan Mode
+- Windsurf Planning Mode  
+- Claude Code Plan Mode
+- Gemini CLI Plan Mode
+- Codex CLI /plan command
+
+Features:
+- PlanningAgent for creating implementation plans
+- Plan and PlanStep dataclasses for plan structure
+- TodoList for tracking progress
+- PlanStorage for persistence
+- ApprovalCallback for plan approval flow
+- Read-only mode for safe research
+
+Usage:
+    from praisonaiagents import Agent, Task, PraisonAIAgents
+    
+    agents = PraisonAIAgents(
+        agents=[agent1, agent2],
+        tasks=[task1, task2],
+        planning=True,           # Enable planning mode
+        planning_llm="gpt-4o-mini"  # Optional: custom LLM
+    )
+    
+    result = agents.start()
+"""
+
+from .plan import Plan, PlanStep
+from .todo import TodoList, TodoItem
+from .storage import PlanStorage
+from .planner import PlanningAgent
+from .approval import ApprovalCallback
+
+# Read-only tools that are allowed in plan mode
+READ_ONLY_TOOLS = [
+    "read_file",
+    "list_directory",
+    "search_codebase",
+    "search_files",
+    "grep_search",
+    "find_files",
+    "web_search",
+    "get_file_content",
+    "list_files",
+    "read_document",
+    "search_web",
+    "fetch_url",
+    "get_context",
+]
+
+# Restricted tools that are blocked in plan mode
+RESTRICTED_TOOLS = [
+    "write_file",
+    "create_file",
+    "delete_file",
+    "execute_command",
+    "run_command",
+    "shell_command",
+    "modify_file",
+    "edit_file",
+    "remove_file",
+    "move_file",
+    "copy_file",
+    "mkdir",
+    "rmdir",
+    "git_commit",
+    "git_push",
+    "npm_install",
+    "pip_install",
+]
+
+__all__ = [
+    "Plan",
+    "PlanStep",
+    "TodoList",
+    "TodoItem",
+    "PlanStorage",
+    "PlanningAgent",
+    "ApprovalCallback",
+    "READ_ONLY_TOOLS",
+    "RESTRICTED_TOOLS",
+]

src/praisonai-agents/praisonaiagents/planning/approval.py

@@ -0,0 +1,274 @@
+"""
+ApprovalCallback for plan approval flow.
+
+Provides approval functionality similar to:
+- Cursor Plan Mode approval
+- Claude Code plan approval
+- Codex CLI /approvals
+
+Features:
+- Auto-approve option
+- Custom approval functions
+- Async support
+- Plan modification during approval
+"""
+
+import asyncio
+import inspect
+import logging
+from typing import Callable, Optional, Union, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .plan import Plan
+
+logger = logging.getLogger(__name__)
+
+
+class ApprovalCallback:
+    """
+    Callback for plan approval flow.
+    
+    Handles the approval process for plans before execution.
+    Can be configured for auto-approval or custom approval logic.
+    
+    Attributes:
+        auto_approve: Whether to automatically approve all plans
+        approve_fn: Custom approval function
+        on_reject: Callback when plan is rejected
+    """
+    
+    def __init__(
+        self,
+        auto_approve: bool = False,
+        approve_fn: Optional[Callable[['Plan'], Union[bool, 'asyncio.Future']]] = None,
+        on_reject: Optional[Callable[['Plan'], None]] = None
+    ):
+        """
+        Initialize ApprovalCallback.
+        
+        Args:
+            auto_approve: If True, automatically approve all plans
+            approve_fn: Custom function to determine approval
+                       Should return True to approve, False to reject
+                       Can be sync or async
+            on_reject: Optional callback when plan is rejected
+        """
+        self.auto_approve = auto_approve
+        self.approve_fn = approve_fn
+        self.on_reject = on_reject
+        
+    def __call__(self, plan: 'Plan') -> bool:
+        """
+        Synchronously check if plan is approved.
+        
+        Args:
+            plan: Plan to check for approval
+            
+        Returns:
+            True if approved, False if rejected
+        """
+        if self.auto_approve:
+            logger.debug(f"Auto-approving plan: {plan.name}")
+            plan.approve()
+            return True
+            
+        if self.approve_fn is not None:
+            # Check if function is async
+            if inspect.iscoroutinefunction(self.approve_fn):
+                # Run async function synchronously
+                loop = asyncio.new_event_loop()
+                try:
+                    result = loop.run_until_complete(self.approve_fn(plan))
+                finally:
+                    loop.close()
+            else:
+                result = self.approve_fn(plan)
+                
+            if result:
+                plan.approve()
+                return True
+            else:
+                if self.on_reject:
+                    self.on_reject(plan)
+                return False
+                
+        # Default: require explicit approval (return False)
+        logger.debug(f"Plan requires explicit approval: {plan.name}")
+        return False
+    
+    async def async_call(self, plan: 'Plan') -> bool:
+        """
+        Asynchronously check if plan is approved.
+        
+        Args:
+            plan: Plan to check for approval
+            
+        Returns:
+            True if approved, False if rejected
+        """
+        if self.auto_approve:
+            logger.debug(f"Auto-approving plan: {plan.name}")
+            plan.approve()
+            return True
+            
+        if self.approve_fn is not None:
+            # Check if function is async
+            if inspect.iscoroutinefunction(self.approve_fn):
+                result = await self.approve_fn(plan)
+            else:
+                result = self.approve_fn(plan)
+                
+            if result:
+                plan.approve()
+                return True
+            else:
+                if self.on_reject:
+                    self.on_reject(plan)
+                return False
+                
+        # Default: require explicit approval (return False)
+        logger.debug(f"Plan requires explicit approval: {plan.name}")
+        return False
+    
+    @staticmethod
+    def console_approval(plan: 'Plan') -> bool:
+        """
+        Interactive console approval.
+        
+        Displays the plan and prompts for approval.
+        
+        Args:
+            plan: Plan to approve
+            
+        Returns:
+            True if user approves, False otherwise
+        """
+        print("\n" + "=" * 60)
+        print("PLAN APPROVAL REQUIRED")
+        print("=" * 60)
+        print(plan.to_markdown())
+        print("=" * 60)
+        
+        while True:
+            response = input("\nApprove this plan? [y/n/e(dit)]: ").strip().lower()
+            
+            if response in ['y', 'yes']:
+                return True
+            elif response in ['n', 'no']:
+                return False
+            elif response in ['e', 'edit']:
+                print("\nPlan editing not implemented in console mode.")
+                print("Please modify the plan programmatically and re-submit.")
+                continue
+            else:
+                print("Please enter 'y' for yes, 'n' for no, or 'e' to edit.")
+                
+    @staticmethod
+    def always_approve(plan: 'Plan') -> bool:
+        """Always approve plans."""
+        return True
+    
+    @staticmethod
+    def always_reject(plan: 'Plan') -> bool:
+        """Always reject plans."""
+        return False
+    
+    @staticmethod
+    def approve_if_small(plan: 'Plan', max_steps: int = 5) -> bool:
+        """
+        Approve plans with few steps automatically.
+        
+        Args:
+            plan: Plan to check
+            max_steps: Maximum number of steps to auto-approve
+            
+        Returns:
+            True if plan has <= max_steps
+        """
+        return len(plan.steps) <= max_steps
+    
+    @staticmethod
+    def approve_if_no_dangerous_tools(plan: 'Plan') -> bool:
+        """
+        Approve plans that don't use dangerous tools.
+        
+        Args:
+            plan: Plan to check
+            
+        Returns:
+            True if no dangerous tools are used
+        """
+        from . import RESTRICTED_TOOLS
+        
+        for step in plan.steps:
+            for tool in step.tools:
+                if tool in RESTRICTED_TOOLS:
+                    logger.warning(f"Plan uses restricted tool: {tool}")
+                    return False
+                    
+        return True
+
+
+class InteractiveApproval:
+    """
+    Interactive approval handler with rich UI support.
+    
+    Provides a more sophisticated approval interface
+    with plan editing capabilities.
+    """
+    
+    def __init__(self, use_rich: bool = True):
+        """
+        Initialize InteractiveApproval.
+        
+        Args:
+            use_rich: Whether to use Rich library for display
+        """
+        self.use_rich = use_rich
+        self._rich_available = False
+        
+        if use_rich:
+            try:
+                from rich.console import Console
+                from rich.panel import Panel
+                from rich.prompt import Confirm
+                self._rich_available = True
+                self._console = Console()
+            except ImportError:
+                self._rich_available = False
+                
+    def __call__(self, plan: 'Plan') -> bool:
+        """
+        Display plan and get approval.
+        
+        Args:
+            plan: Plan to approve
+            
+        Returns:
+            True if approved
+        """
+        if self._rich_available:
+            return self._rich_approval(plan)
+        else:
+            return ApprovalCallback.console_approval(plan)
+            
+    def _rich_approval(self, plan: 'Plan') -> bool:
+        """Rich library approval display."""
+        from rich.panel import Panel
+        from rich.prompt import Confirm
+        from rich.markdown import Markdown
+        
+        # Display plan
+        md = Markdown(plan.to_markdown())
+        self._console.print(Panel(md, title="[bold blue]Plan Approval Required[/bold blue]"))
+        
+        # Get approval
+        approved = Confirm.ask("Approve this plan?")
+        
+        return approved
+    
+    async def async_call(self, plan: 'Plan') -> bool:
+        """Async version of approval."""
+        # For now, just call sync version
+        # Could be extended for async UI frameworks
+        return self(plan)

src/praisonai-agents/praisonaiagents/planning/plan.py

@@ -0,0 +1,364 @@
+"""
+Plan and PlanStep dataclasses for Planning Mode.
+
+Provides structured representation of implementation plans
+with support for:
+- Step dependencies
+- Status tracking
+- Markdown serialization
+- Progress calculation
+"""
+
+import uuid
+import re
+import yaml
+import logging
+from datetime import datetime
+from dataclasses import dataclass, field
+from typing import List, Optional, Dict, Any, Literal
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class PlanStep:
+    """
+    A single step in an implementation plan.
+    
+    Attributes:
+        id: Unique identifier for the step
+        description: What this step does
+        agent: Name of the agent to execute this step
+        tools: List of tools needed for this step
+        dependencies: List of step IDs that must complete first
+        status: Current status (pending, in_progress, completed, skipped)
+        estimated_tokens: Estimated token usage for this step
+    """
+    description: str
+    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
+    agent: Optional[str] = None
+    tools: List[str] = field(default_factory=list)
+    dependencies: List[str] = field(default_factory=list)
+    status: Literal["pending", "in_progress", "completed", "skipped"] = "pending"
+    estimated_tokens: int = 0
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert PlanStep to dictionary."""
+        return {
+            "id": self.id,
+            "description": self.description,
+            "agent": self.agent,
+            "tools": self.tools,
+            "dependencies": self.dependencies,
+            "status": self.status,
+            "estimated_tokens": self.estimated_tokens
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'PlanStep':
+        """Create PlanStep from dictionary."""
+        return cls(
+            id=data.get("id", str(uuid.uuid4())[:8]),
+            description=data.get("description", ""),
+            agent=data.get("agent"),
+            tools=data.get("tools", []),
+            dependencies=data.get("dependencies", []),
+            status=data.get("status", "pending"),
+            estimated_tokens=data.get("estimated_tokens", 0)
+        )
+    
+    def mark_complete(self) -> None:
+        """Mark this step as completed."""
+        self.status = "completed"
+        
+    def mark_in_progress(self) -> None:
+        """Mark this step as in progress."""
+        self.status = "in_progress"
+        
+    def mark_skipped(self) -> None:
+        """Mark this step as skipped."""
+        self.status = "skipped"
+        
+    def is_ready(self, completed_steps: List[str]) -> bool:
+        """
+        Check if this step is ready to execute.
+        
+        Args:
+            completed_steps: List of completed step IDs
+            
+        Returns:
+            True if all dependencies are satisfied
+        """
+        if self.status != "pending":
+            return False
+        return all(dep in completed_steps for dep in self.dependencies)
+
+
+@dataclass
+class Plan:
+    """
+    An implementation plan with multiple steps.
+    
+    Attributes:
+        id: Unique identifier for the plan
+        name: Human-readable name
+        description: Detailed description of what the plan accomplishes
+        steps: List of PlanStep objects
+        status: Current status (draft, approved, executing, completed)
+        created_at: When the plan was created
+        approved_at: When the plan was approved (if applicable)
+        file_path: Path where the plan is stored (if saved)
+    """
+    name: str
+    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
+    description: str = ""
+    steps: List[PlanStep] = field(default_factory=list)
+    status: Literal["draft", "approved", "executing", "completed", "cancelled"] = "draft"
+    created_at: datetime = field(default_factory=datetime.now)
+    approved_at: Optional[datetime] = None
+    file_path: Optional[str] = None
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert Plan to dictionary."""
+        return {
+            "id": self.id,
+            "name": self.name,
+            "description": self.description,
+            "steps": [step.to_dict() for step in self.steps],
+            "status": self.status,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "approved_at": self.approved_at.isoformat() if self.approved_at else None,
+            "file_path": self.file_path
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'Plan':
+        """Create Plan from dictionary."""
+        steps = [
+            PlanStep.from_dict(s) if isinstance(s, dict) else s
+            for s in data.get("steps", [])
+        ]
+        
+        created_at = data.get("created_at")
+        if isinstance(created_at, str):
+            created_at = datetime.fromisoformat(created_at)
+        elif created_at is None:
+            created_at = datetime.now()
+            
+        approved_at = data.get("approved_at")
+        if isinstance(approved_at, str):
+            approved_at = datetime.fromisoformat(approved_at)
+            
+        return cls(
+            id=data.get("id", str(uuid.uuid4())[:8]),
+            name=data.get("name", "Untitled Plan"),
+            description=data.get("description", ""),
+            steps=steps,
+            status=data.get("status", "draft"),
+            created_at=created_at,
+            approved_at=approved_at,
+            file_path=data.get("file_path")
+        )
+    
+    def to_markdown(self) -> str:
+        """
+        Convert Plan to markdown format.
+        
+        Returns:
+            Markdown string representation of the plan
+        """
+        # YAML frontmatter
+        frontmatter = {
+            "id": self.id,
+            "name": self.name,
+            "status": self.status,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "approved_at": self.approved_at.isoformat() if self.approved_at else None
+        }
+        
+        lines = [
+            "---",
+            yaml.dump(frontmatter, default_flow_style=False).strip(),
+            "---",
+            "",
+            f"# {self.name}",
+            "",
+        ]
+        
+        if self.description:
+            lines.extend([self.description, ""])
+            
+        lines.extend(["## Steps", ""])
+        
+        status_icons = {
+            "pending": "⬜",
+            "in_progress": "🔄",
+            "completed": "✅",
+            "skipped": "⏭️"
+        }
+        
+        for i, step in enumerate(self.steps, 1):
+            icon = status_icons.get(step.status, "⬜")
+            lines.append(f"{i}. {icon} {step.description}")
+            
+            if step.agent:
+                lines.append(f"   - Agent: {step.agent}")
+            if step.tools:
+                lines.append(f"   - Tools: {', '.join(step.tools)}")
+            if step.dependencies:
+                lines.append(f"   - Depends on: {', '.join(step.dependencies)}")
+            lines.append("")
+            
+        return "\n".join(lines)
+    
+    @classmethod
+    def from_markdown(cls, markdown: str) -> 'Plan':
+        """
+        Create Plan from markdown format.
+        
+        Args:
+            markdown: Markdown string with YAML frontmatter
+            
+        Returns:
+            Plan instance
+        """
+        # Extract YAML frontmatter
+        frontmatter_match = re.match(r'^---\n(.*?)\n---', markdown, re.DOTALL)
+        
+        data = {}
+        if frontmatter_match:
+            try:
+                data = yaml.safe_load(frontmatter_match.group(1))
+            except yaml.YAMLError:
+                logger.warning("Failed to parse YAML frontmatter")
+                
+        # Extract steps from markdown
+        steps = []
+        # Pattern to capture status icon and description
+        step_pattern = r'^\d+\.\s*([⬜🔄✅⏭️]?)\s*(.+?)$'
+        
+        # Map icons to status
+        icon_to_status = {
+            "⬜": "pending",
+            "🔄": "in_progress",
+            "✅": "completed",
+            "⏭️": "skipped",
+            "": "pending"
+        }
+        
+        for match in re.finditer(step_pattern, markdown, re.MULTILINE):
+            icon = match.group(1).strip()
+            description = match.group(2).strip()
+            status = icon_to_status.get(icon, "pending")
+            steps.append(PlanStep(description=description, status=status))
+            
+        return cls(
+            id=data.get("id", str(uuid.uuid4())[:8]),
+            name=data.get("name", "Untitled Plan"),
+            description=data.get("description", ""),
+            steps=steps,
+            status=data.get("status", "draft")
+        )
+    
+    def add_step(self, step: PlanStep) -> None:
+        """Add a step to the plan."""
+        self.steps.append(step)
+        
+    def remove_step(self, step_id: str) -> bool:
+        """
+        Remove a step from the plan.
+        
+        Args:
+            step_id: ID of the step to remove
+            
+        Returns:
+            True if step was removed, False if not found
+        """
+        for i, step in enumerate(self.steps):
+            if step.id == step_id:
+                self.steps.pop(i)
+                return True
+        return False
+    
+    def get_step(self, step_id: str) -> Optional[PlanStep]:
+        """
+        Get a step by ID.
+        
+        Args:
+            step_id: ID of the step to find
+            
+        Returns:
+            PlanStep if found, None otherwise
+        """
+        for step in self.steps:
+            if step.id == step_id:
+                return step
+        return None
+    
+    def approve(self) -> None:
+        """Approve the plan for execution."""
+        self.status = "approved"
+        self.approved_at = datetime.now()
+        
+    def start_execution(self) -> None:
+        """Mark the plan as executing."""
+        self.status = "executing"
+        
+    def complete(self) -> None:
+        """Mark the plan as completed."""
+        self.status = "completed"
+        
+    def cancel(self) -> None:
+        """Cancel the plan."""
+        self.status = "cancelled"
+    
+    @property
+    def progress(self) -> float:
+        """
+        Calculate plan progress.
+        
+        Returns:
+            Float between 0.0 and 1.0 representing completion percentage
+        """
+        if not self.steps:
+            return 1.0
+            
+        completed = sum(1 for s in self.steps if s.status == "completed")
+        return completed / len(self.steps)
+    
+    @property
+    def is_complete(self) -> bool:
+        """Check if all steps are completed."""
+        return all(s.status == "completed" for s in self.steps)
+    
+    @property
+    def completed_step_ids(self) -> List[str]:
+        """Get list of completed step IDs."""
+        return [s.id for s in self.steps if s.status == "completed"]
+    
+    def get_next_steps(self) -> List[PlanStep]:
+        """
+        Get steps that are ready to execute.
+        
+        Returns:
+            List of steps with satisfied dependencies
+        """
+        completed = self.completed_step_ids
+        return [s for s in self.steps if s.is_ready(completed)]
+    
+    def update_step_status(self, step_id: str, status: str) -> bool:
+        """
+        Update the status of a step.
+        
+        Args:
+            step_id: ID of the step to update
+            status: New status
+            
+        Returns:
+            True if step was updated, False if not found
+        """
+        step = self.get_step(step_id)
+        if step:
+            step.status = status
+            return True
+        return False

src/praisonai-agents/praisonaiagents/planning/planner.py

@@ -0,0 +1,488 @@
+"""
+PlanningAgent for creating implementation plans.
+
+Provides planning functionality similar to:
+- CrewAI AgentPlanner
+- Cursor Plan Mode
+- Claude Code Plan Mode
+
+Features:
+- Analyze requests and create detailed plans
+- Read-only mode for safe research
+- Plan refinement based on feedback
+- Context analysis
+"""
+
+import json
+import logging
+from typing import List, Optional, Dict, Any, TYPE_CHECKING
+
+from .plan import Plan, PlanStep
+
+# Define READ_ONLY_TOOLS locally to avoid circular import
+READ_ONLY_TOOLS = [
+    "read_file",
+    "list_directory",
+    "search_codebase",
+    "search_files",
+    "grep_search",
+    "find_files",
+    "web_search",
+    "get_file_content",
+    "list_files",
+    "read_document",
+    "search_web",
+    "fetch_url",
+    "get_context",
+]
+
+if TYPE_CHECKING:
+    from ..agent.agent import Agent
+    from ..task.task import Task
+
+logger = logging.getLogger(__name__)
+
+
+class PlanningAgent:
+    """
+    Agent specialized in creating implementation plans.
+    
+    The PlanningAgent analyzes requests, researches the codebase,
+    and creates detailed step-by-step plans before execution.
+    
+    Attributes:
+        llm_model: LLM model to use for planning
+        read_only: Whether to restrict to read-only tools
+        allowed_tools: List of tools available in planning mode
+        verbose: Verbosity level
+    """
+    
+    PLANNING_PROMPT = '''You are an expert planning specialist. Your task is to analyze the request and create a detailed, actionable implementation plan.
+
+## Request
+{request}
+
+## Available Agents
+{agents_info}
+
+## Available Tasks (if any)
+{tasks_info}
+
+## Context
+{context}
+
+## Instructions
+1. Analyze the request thoroughly
+2. Break it down into clear, actionable steps
+3. Assign appropriate agents to each step
+4. Identify dependencies between steps
+5. Consider potential risks and edge cases
+
+## Output Format
+Respond with a JSON object in this exact format:
+{{
+    "name": "Brief plan name",
+    "description": "Detailed description of what this plan accomplishes",
+    "steps": [
+        {{
+            "description": "Clear description of what this step does",
+            "agent": "Name of the agent to execute this step (or null)",
+            "tools": ["list", "of", "tools", "needed"],
+            "dependencies": ["list of step IDs this depends on"]
+        }}
+    ]
+}}
+
+Important:
+- Each step should be atomic and clearly defined
+- Dependencies should reference step indices (e.g., "step_0", "step_1")
+- Be specific about what each step accomplishes
+- Consider the order of operations carefully'''
+
+    REFINEMENT_PROMPT = '''You are refining an existing implementation plan based on feedback.
+
+## Original Plan
+{original_plan}
+
+## Feedback
+{feedback}
+
+## Instructions
+1. Review the original plan
+2. Apply the feedback to improve the plan
+3. Maintain the same JSON format
+4. Keep steps that are still valid
+5. Add, modify, or remove steps as needed
+
+## Output Format
+Respond with a JSON object in the same format as the original plan:
+{{
+    "name": "Updated plan name",
+    "description": "Updated description",
+    "steps": [...]
+}}'''
+
+    ANALYSIS_PROMPT = '''Analyze the following context and provide insights for planning:
+
+## Context
+{context}
+
+## Instructions
+1. Identify the key components and their relationships
+2. Note any patterns or conventions
+3. Highlight potential challenges
+4. Suggest best practices to follow
+
+Provide a concise analysis that will help in creating an implementation plan.'''
+
+    def __init__(
+        self,
+        llm: str = "gpt-4o-mini",
+        read_only: bool = True,
+        verbose: int = 0
+    ):
+        """
+        Initialize PlanningAgent.
+        
+        Args:
+            llm: LLM model to use for planning
+            read_only: Whether to restrict to read-only tools
+            verbose: Verbosity level
+        """
+        self.llm_model = llm
+        self.read_only = read_only
+        self.verbose = verbose
+        self.allowed_tools = READ_ONLY_TOOLS.copy() if read_only else []
+        self._llm = None
+        
+    def _get_llm(self):
+        """Lazy load LLM instance."""
+        if self._llm is None:
+            from ..llm import LLM
+            self._llm = LLM(model=self.llm_model)
+        return self._llm
+    
+    def _format_agents_info(self, agents: List['Agent']) -> str:
+        """
+        Format agent information for the prompt.
+        
+        Args:
+            agents: List of Agent instances
+            
+        Returns:
+            Formatted string describing agents
+        """
+        if not agents:
+            return "No agents specified"
+            
+        lines = []
+        for agent in agents:
+            name = getattr(agent, 'name', 'Unnamed')
+            role = getattr(agent, 'role', 'General')
+            goal = getattr(agent, 'goal', '')
+            
+            line = f"- **{name}** ({role})"
+            if goal:
+                line += f": {goal}"
+            lines.append(line)
+            
+        return "\n".join(lines)
+    
+    def _format_tasks_info(self, tasks: Optional[List['Task']]) -> str:
+        """
+        Format task information for the prompt.
+        
+        Args:
+            tasks: List of Task instances
+            
+        Returns:
+            Formatted string describing tasks
+        """
+        if not tasks:
+            return "No predefined tasks"
+            
+        lines = []
+        for task in tasks:
+            name = getattr(task, 'name', None) or getattr(task, 'description', 'Unnamed')[:50]
+            desc = getattr(task, 'description', '')[:100]
+            
+            lines.append(f"- {name}: {desc}")
+            
+        return "\n".join(lines)
+    
+    def _parse_plan_response(self, response: str) -> Plan:
+        """
+        Parse LLM response into a Plan object.
+        
+        Args:
+            response: JSON string from LLM
+            
+        Returns:
+            Plan instance
+        """
+        try:
+            # Try to extract JSON from response
+            # Handle cases where LLM includes extra text
+            json_start = response.find('{')
+            json_end = response.rfind('}') + 1
+            
+            if json_start >= 0 and json_end > json_start:
+                json_str = response[json_start:json_end]
+                data = json.loads(json_str)
+            else:
+                raise ValueError("No JSON found in response")
+                
+            # Create steps
+            steps = []
+            for i, step_data in enumerate(data.get("steps", [])):
+                step = PlanStep(
+                    id=f"step_{i}",
+                    description=step_data.get("description", ""),
+                    agent=step_data.get("agent"),
+                    tools=step_data.get("tools", []),
+                    dependencies=step_data.get("dependencies", [])
+                )
+                steps.append(step)
+                
+            return Plan(
+                name=data.get("name", "Implementation Plan"),
+                description=data.get("description", ""),
+                steps=steps
+            )
+            
+        except (json.JSONDecodeError, ValueError) as e:
+            logger.warning(f"Failed to parse plan response: {e}")
+            # Return a simple plan with the response as a single step
+            return Plan(
+                name="Implementation Plan",
+                description="Auto-generated plan",
+                steps=[PlanStep(description=response[:500])]
+            )
+    
+    async def create_plan(
+        self,
+        request: str,
+        agents: List['Agent'],
+        tasks: Optional[List['Task']] = None,
+        context: Optional[str] = None
+    ) -> Plan:
+        """
+        Create an implementation plan for the given request.
+        
+        Args:
+            request: The user's request or goal
+            agents: List of available agents
+            tasks: Optional list of predefined tasks
+            context: Optional additional context
+            
+        Returns:
+            Plan instance with steps
+        """
+        llm = self._get_llm()
+        
+        # Format the prompt
+        prompt = self.PLANNING_PROMPT.format(
+            request=request,
+            agents_info=self._format_agents_info(agents),
+            tasks_info=self._format_tasks_info(tasks),
+            context=context or "No additional context provided"
+        )
+        
+        if self.verbose >= 2:
+            logger.info(f"Creating plan for: {request[:100]}...")
+            
+        # Get LLM response
+        response = await llm.get_response_async(prompt)
+        
+        # Parse response into Plan
+        plan = self._parse_plan_response(response)
+        
+        if self.verbose >= 1:
+            logger.info(f"Created plan '{plan.name}' with {len(plan.steps)} steps")
+            
+        return plan
+    
+    def create_plan_sync(
+        self,
+        request: str,
+        agents: List['Agent'],
+        tasks: Optional[List['Task']] = None,
+        context: Optional[str] = None
+    ) -> Plan:
+        """
+        Synchronous version of create_plan.
+        
+        Args:
+            request: The user's request or goal
+            agents: List of available agents
+            tasks: Optional list of predefined tasks
+            context: Optional additional context
+            
+        Returns:
+            Plan instance with steps
+        """
+        llm = self._get_llm()
+        
+        # Format the prompt
+        prompt = self.PLANNING_PROMPT.format(
+            request=request,
+            agents_info=self._format_agents_info(agents),
+            tasks_info=self._format_tasks_info(tasks),
+            context=context or "No additional context provided"
+        )
+        
+        if self.verbose >= 2:
+            logger.info(f"Creating plan for: {request[:100]}...")
+            
+        # Get LLM response
+        response = llm.get_response(prompt)
+        
+        # Parse response into Plan
+        plan = self._parse_plan_response(response)
+        
+        if self.verbose >= 1:
+            logger.info(f"Created plan '{plan.name}' with {len(plan.steps)} steps")
+            
+        return plan
+    
+    async def refine_plan(
+        self,
+        plan: Plan,
+        feedback: str
+    ) -> Plan:
+        """
+        Refine an existing plan based on feedback.
+        
+        Args:
+            plan: Original plan to refine
+            feedback: User feedback for refinement
+            
+        Returns:
+            Refined Plan instance
+        """
+        llm = self._get_llm()
+        
+        # Format the prompt
+        prompt = self.REFINEMENT_PROMPT.format(
+            original_plan=json.dumps(plan.to_dict(), indent=2),
+            feedback=feedback
+        )
+        
+        if self.verbose >= 2:
+            logger.info(f"Refining plan based on feedback: {feedback[:100]}...")
+            
+        # Get LLM response
+        response = await llm.get_response_async(prompt)
+        
+        # Parse response into Plan
+        refined = self._parse_plan_response(response)
+        
+        # Preserve original ID
+        refined.id = plan.id
+        
+        if self.verbose >= 1:
+            logger.info(f"Refined plan '{refined.name}' with {len(refined.steps)} steps")
+            
+        return refined
+    
+    def refine_plan_sync(
+        self,
+        plan: Plan,
+        feedback: str
+    ) -> Plan:
+        """
+        Synchronous version of refine_plan.
+        
+        Args:
+            plan: Original plan to refine
+            feedback: User feedback for refinement
+            
+        Returns:
+            Refined Plan instance
+        """
+        llm = self._get_llm()
+        
+        # Format the prompt
+        prompt = self.REFINEMENT_PROMPT.format(
+            original_plan=json.dumps(plan.to_dict(), indent=2),
+            feedback=feedback
+        )
+        
+        if self.verbose >= 2:
+            logger.info(f"Refining plan based on feedback: {feedback[:100]}...")
+            
+        # Get LLM response
+        response = llm.get_response(prompt)
+        
+        # Parse response into Plan
+        refined = self._parse_plan_response(response)
+        
+        # Preserve original ID
+        refined.id = plan.id
+        
+        if self.verbose >= 1:
+            logger.info(f"Refined plan '{refined.name}' with {len(refined.steps)} steps")
+            
+        return refined
+    
+    async def analyze_context(
+        self,
+        context: str
+    ) -> str:
+        """
+        Analyze context and provide insights for planning.
+        
+        Args:
+            context: Context to analyze (e.g., codebase summary)
+            
+        Returns:
+            Analysis string
+        """
+        llm = self._get_llm()
+        
+        prompt = self.ANALYSIS_PROMPT.format(context=context)
+        
+        if self.verbose >= 2:
+            logger.info("Analyzing context...")
+            
+        response = await llm.get_response_async(prompt)
+        
+        return response
+    
+    def analyze_context_sync(
+        self,
+        context: str
+    ) -> str:
+        """
+        Synchronous version of analyze_context.
+        
+        Args:
+            context: Context to analyze
+            
+        Returns:
+            Analysis string
+        """
+        llm = self._get_llm()
+        
+        prompt = self.ANALYSIS_PROMPT.format(context=context)
+        
+        if self.verbose >= 2:
+            logger.info("Analyzing context...")
+            
+        response = llm.get_response(prompt)
+        
+        return response
+    
+    def is_tool_allowed(self, tool_name: str) -> bool:
+        """
+        Check if a tool is allowed in planning mode.
+        
+        Args:
+            tool_name: Name of the tool to check
+            
+        Returns:
+            True if tool is allowed
+        """
+        if not self.read_only:
+            return True
+        return tool_name in self.allowed_tools

src/praisonai-agents/praisonaiagents/planning/storage.py

@@ -0,0 +1,325 @@
+"""
+PlanStorage for persisting plans and todo lists.
+
+Provides storage functionality similar to:
+- Cursor .cursor/plans/
+- Windsurf ~/.codeium/windsurf/brain/
+- Codex ~/.codex/sessions/
+
+Features:
+- Save/load plans as markdown files
+- Save/load todo lists as JSON
+- Auto-discovery of existing plans
+- Plan versioning
+"""
+
+import os
+import json
+import logging
+from pathlib import Path
+from datetime import datetime
+from typing import List, Optional, Dict, Any
+
+from .plan import Plan
+from .todo import TodoList
+
+logger = logging.getLogger(__name__)
+
+
+class PlanStorage:
+    """
+    Storage manager for plans and todo lists.
+    
+    Attributes:
+        base_path: Base directory for storage
+        plans_dir: Directory for plan files
+        todos_dir: Directory for todo files
+    """
+    
+    DEFAULT_BASE_PATH = ".praison"
+    
+    def __init__(self, base_path: Optional[str] = None):
+        """
+        Initialize PlanStorage.
+        
+        Args:
+            base_path: Base directory for storage. Defaults to .praison/
+        """
+        self.base_path = base_path or self.DEFAULT_BASE_PATH
+        self.plans_dir = os.path.join(self.base_path, "plans")
+        self.todos_dir = os.path.join(self.base_path, "todos")
+        
+        # Create directories if they don't exist
+        self._ensure_directories()
+        
+    def _ensure_directories(self) -> None:
+        """Create storage directories if they don't exist."""
+        os.makedirs(self.plans_dir, exist_ok=True)
+        os.makedirs(self.todos_dir, exist_ok=True)
+        
+    def _get_plan_filename(self, plan: Plan) -> str:
+        """
+        Generate filename for a plan.
+        
+        Args:
+            plan: Plan to generate filename for
+            
+        Returns:
+            Filename string
+        """
+        date_str = plan.created_at.strftime("%Y-%m-%d") if plan.created_at else datetime.now().strftime("%Y-%m-%d")
+        safe_name = "".join(c if c.isalnum() or c in "-_" else "_" for c in plan.name[:30])
+        return f"plan_{date_str}_{plan.id}_{safe_name}.md"
+    
+    def _get_plan_path(self, plan_id: str) -> Optional[str]:
+        """
+        Find the path for a plan by ID.
+        
+        Args:
+            plan_id: ID of the plan to find
+            
+        Returns:
+            Path to the plan file, or None if not found
+        """
+        for filename in os.listdir(self.plans_dir):
+            if plan_id in filename and filename.endswith(".md"):
+                return os.path.join(self.plans_dir, filename)
+        return None
+    
+    def save_plan(self, plan: Plan, filename: Optional[str] = None) -> str:
+        """
+        Save a plan to storage.
+        
+        Args:
+            plan: Plan to save
+            filename: Optional custom filename
+            
+        Returns:
+            Path to the saved file
+        """
+        if filename is None:
+            filename = self._get_plan_filename(plan)
+            
+        path = os.path.join(self.plans_dir, filename)
+        
+        # Convert plan to markdown
+        markdown = plan.to_markdown()
+        
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(markdown)
+            
+        plan.file_path = path
+        logger.debug(f"Saved plan to {path}")
+        
+        return path
+    
+    def load_plan(self, plan_id: str) -> Optional[Plan]:
+        """
+        Load a plan from storage.
+        
+        Args:
+            plan_id: ID of the plan to load
+            
+        Returns:
+            Plan instance, or None if not found
+        """
+        path = self._get_plan_path(plan_id)
+        
+        if path is None or not os.path.exists(path):
+            logger.debug(f"Plan {plan_id} not found")
+            return None
+            
+        with open(path, "r", encoding="utf-8") as f:
+            markdown = f.read()
+            
+        plan = Plan.from_markdown(markdown)
+        plan.file_path = path
+        
+        logger.debug(f"Loaded plan from {path}")
+        return plan
+    
+    def load_plan_from_file(self, path: str) -> Optional[Plan]:
+        """
+        Load a plan from a specific file path.
+        
+        Args:
+            path: Path to the plan file
+            
+        Returns:
+            Plan instance, or None if file doesn't exist
+        """
+        if not os.path.exists(path):
+            return None
+            
+        with open(path, "r", encoding="utf-8") as f:
+            markdown = f.read()
+            
+        plan = Plan.from_markdown(markdown)
+        plan.file_path = path
+        
+        return plan
+    
+    def list_plans(self) -> List[Dict[str, Any]]:
+        """
+        List all saved plans.
+        
+        Returns:
+            List of plan metadata dictionaries
+        """
+        plans = []
+        
+        for filename in os.listdir(self.plans_dir):
+            if filename.endswith(".md"):
+                path = os.path.join(self.plans_dir, filename)
+                
+                # Try to load and get metadata
+                try:
+                    plan = self.load_plan_from_file(path)
+                    if plan:
+                        plans.append({
+                            "id": plan.id,
+                            "name": plan.name,
+                            "status": plan.status,
+                            "created_at": plan.created_at.isoformat() if plan.created_at else None,
+                            "file_path": path
+                        })
+                except Exception as e:
+                    logger.warning(f"Failed to load plan from {path}: {e}")
+                    
+        # Sort by creation date (newest first)
+        plans.sort(key=lambda p: p.get("created_at", ""), reverse=True)
+        
+        return plans
+    
+    def delete_plan(self, plan_id: str) -> bool:
+        """
+        Delete a plan from storage.
+        
+        Args:
+            plan_id: ID of the plan to delete
+            
+        Returns:
+            True if deleted, False if not found
+        """
+        path = self._get_plan_path(plan_id)
+        
+        if path and os.path.exists(path):
+            os.remove(path)
+            logger.debug(f"Deleted plan {plan_id}")
+            return True
+            
+        return False
+    
+    def save_todo(self, todo: TodoList, name: str = "current") -> str:
+        """
+        Save a todo list to storage.
+        
+        Args:
+            todo: TodoList to save
+            name: Name for the todo file (without extension)
+            
+        Returns:
+            Path to the saved file
+        """
+        filename = f"{name}.json"
+        path = os.path.join(self.todos_dir, filename)
+        
+        with open(path, "w", encoding="utf-8") as f:
+            f.write(todo.to_json())
+            
+        logger.debug(f"Saved todo list to {path}")
+        return path
+    
+    def load_todo(self, name: str = "current") -> Optional[TodoList]:
+        """
+        Load a todo list from storage.
+        
+        Args:
+            name: Name of the todo file (without extension)
+            
+        Returns:
+            TodoList instance, or None if not found
+        """
+        filename = f"{name}.json"
+        path = os.path.join(self.todos_dir, filename)
+        
+        if not os.path.exists(path):
+            return None
+            
+        with open(path, "r", encoding="utf-8") as f:
+            json_str = f.read()
+            
+        logger.debug(f"Loaded todo list from {path}")
+        return TodoList.from_json(json_str)
+    
+    def list_todos(self) -> List[str]:
+        """
+        List all saved todo lists.
+        
+        Returns:
+            List of todo list names
+        """
+        todos = []
+        
+        for filename in os.listdir(self.todos_dir):
+            if filename.endswith(".json"):
+                name = filename[:-5]  # Remove .json extension
+                todos.append(name)
+                
+        return todos
+    
+    def delete_todo(self, name: str) -> bool:
+        """
+        Delete a todo list from storage.
+        
+        Args:
+            name: Name of the todo file (without extension)
+            
+        Returns:
+            True if deleted, False if not found
+        """
+        filename = f"{name}.json"
+        path = os.path.join(self.todos_dir, filename)
+        
+        if os.path.exists(path):
+            os.remove(path)
+            logger.debug(f"Deleted todo list {name}")
+            return True
+            
+        return False
+    
+    def get_latest_plan(self) -> Optional[Plan]:
+        """
+        Get the most recently created plan.
+        
+        Returns:
+            Most recent Plan, or None if no plans exist
+        """
+        plans = self.list_plans()
+        
+        if not plans:
+            return None
+            
+        return self.load_plan(plans[0]["id"])
+    
+    def cleanup_old_plans(self, keep_count: int = 10) -> int:
+        """
+        Remove old plans, keeping only the most recent ones.
+        
+        Args:
+            keep_count: Number of plans to keep
+            
+        Returns:
+            Number of plans deleted
+        """
+        plans = self.list_plans()
+        
+        if len(plans) <= keep_count:
+            return 0
+            
+        deleted = 0
+        for plan in plans[keep_count:]:
+            if self.delete_plan(plan["id"]):
+                deleted += 1
+                
+        return deleted

src/praisonai-agents/praisonaiagents/planning/todo.py

@@ -0,0 +1,415 @@
+"""
+TodoList and TodoItem classes for Planning Mode.
+
+Provides task tracking functionality similar to:
+- Cursor Agent To-Dos
+- Windsurf Todo Lists
+- Claude Code TodoRead/TodoWrite
+
+Features:
+- Create todo lists from plans
+- Track item completion
+- Dependencies between items
+- Markdown and JSON serialization
+"""
+
+import uuid
+import json
+import logging
+from dataclasses import dataclass, field
+from typing import List, Optional, Dict, Any, Literal, Union, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .plan import Plan
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class TodoItem:
+    """
+    A single todo item.
+    
+    Attributes:
+        id: Unique identifier
+        description: What needs to be done
+        status: Current status (pending, in_progress, completed)
+        dependencies: List of item IDs that must complete first
+        agent: Name of the agent responsible
+        priority: Priority level (low, medium, high)
+        notes: Additional notes
+    """
+    description: str
+    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
+    status: Literal["pending", "in_progress", "completed"] = "pending"
+    dependencies: List[str] = field(default_factory=list)
+    agent: Optional[str] = None
+    priority: Literal["low", "medium", "high"] = "medium"
+    notes: str = ""
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert TodoItem to dictionary."""
+        return {
+            "id": self.id,
+            "description": self.description,
+            "status": self.status,
+            "dependencies": self.dependencies,
+            "agent": self.agent,
+            "priority": self.priority,
+            "notes": self.notes
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'TodoItem':
+        """Create TodoItem from dictionary."""
+        return cls(
+            id=data.get("id", str(uuid.uuid4())[:8]),
+            description=data.get("description", ""),
+            status=data.get("status", "pending"),
+            dependencies=data.get("dependencies", []),
+            agent=data.get("agent"),
+            priority=data.get("priority", "medium"),
+            notes=data.get("notes", "")
+        )
+    
+    def complete(self) -> None:
+        """Mark this item as completed."""
+        self.status = "completed"
+        
+    def start(self) -> None:
+        """Mark this item as in progress."""
+        self.status = "in_progress"
+        
+    def reset(self) -> None:
+        """Reset this item to pending."""
+        self.status = "pending"
+        
+    def is_ready(self, completed_ids: List[str]) -> bool:
+        """
+        Check if this item is ready to start.
+        
+        Args:
+            completed_ids: List of completed item IDs
+            
+        Returns:
+            True if all dependencies are satisfied
+        """
+        if self.status != "pending":
+            return False
+        return all(dep in completed_ids for dep in self.dependencies)
+
+
+@dataclass
+class TodoList:
+    """
+    A list of todo items with tracking capabilities.
+    
+    Attributes:
+        items: List of TodoItem objects
+        auto_update: Whether to auto-update from plan changes
+        name: Optional name for the todo list
+    """
+    items: List[TodoItem] = field(default_factory=list)
+    auto_update: bool = True
+    name: str = "Todo List"
+    
+    def add(self, item: Union[TodoItem, str]) -> TodoItem:
+        """
+        Add an item to the list.
+        
+        Args:
+            item: TodoItem or description string
+            
+        Returns:
+            The added TodoItem
+        """
+        if isinstance(item, str):
+            item = TodoItem(description=item)
+        self.items.append(item)
+        return item
+    
+    def remove(self, item_id: str) -> bool:
+        """
+        Remove an item from the list.
+        
+        Args:
+            item_id: ID of the item to remove
+            
+        Returns:
+            True if item was removed, False if not found
+        """
+        for i, item in enumerate(self.items):
+            if item.id == item_id:
+                self.items.pop(i)
+                return True
+        return False
+    
+    def get(self, item_id: str) -> Optional[TodoItem]:
+        """
+        Get an item by ID.
+        
+        Args:
+            item_id: ID of the item to find
+            
+        Returns:
+            TodoItem if found, None otherwise
+        """
+        for item in self.items:
+            if item.id == item_id:
+                return item
+        return None
+    
+    def complete(self, item_id: str) -> bool:
+        """
+        Mark an item as completed.
+        
+        Args:
+            item_id: ID of the item to complete
+            
+        Returns:
+            True if item was completed, False if not found
+        """
+        item = self.get(item_id)
+        if item:
+            item.complete()
+            return True
+        return False
+    
+    def start(self, item_id: str) -> bool:
+        """
+        Mark an item as in progress.
+        
+        Args:
+            item_id: ID of the item to start
+            
+        Returns:
+            True if item was started, False if not found
+        """
+        item = self.get(item_id)
+        if item:
+            item.start()
+            return True
+        return False
+    
+    @property
+    def pending(self) -> List[TodoItem]:
+        """Get all pending items."""
+        return [item for item in self.items if item.status == "pending"]
+    
+    @property
+    def in_progress(self) -> List[TodoItem]:
+        """Get all in-progress items."""
+        return [item for item in self.items if item.status == "in_progress"]
+    
+    @property
+    def completed(self) -> List[TodoItem]:
+        """Get all completed items."""
+        return [item for item in self.items if item.status == "completed"]
+    
+    @property
+    def progress(self) -> float:
+        """
+        Calculate completion progress.
+        
+        Returns:
+            Float between 0.0 and 1.0
+        """
+        if not self.items:
+            return 1.0
+        return len(self.completed) / len(self.items)
+    
+    @property
+    def is_complete(self) -> bool:
+        """Check if all items are completed."""
+        return all(item.status == "completed" for item in self.items)
+    
+    @property
+    def completed_ids(self) -> List[str]:
+        """Get list of completed item IDs."""
+        return [item.id for item in self.completed]
+    
+    def get_ready_items(self) -> List[TodoItem]:
+        """
+        Get items that are ready to start.
+        
+        Returns:
+            List of items with satisfied dependencies
+        """
+        completed = self.completed_ids
+        return [item for item in self.items if item.is_ready(completed)]
+    
+    def to_markdown(self) -> str:
+        """
+        Convert TodoList to markdown format.
+        
+        Returns:
+            Markdown string with checkbox format
+        """
+        lines = [f"# {self.name}", ""]
+        
+        for item in self.items:
+            checkbox = "[x]" if item.status == "completed" else "[ ]"
+            lines.append(f"- {checkbox} {item.description}")
+            
+            if item.agent:
+                lines.append(f"  - Agent: {item.agent}")
+            if item.dependencies:
+                lines.append(f"  - Depends on: {', '.join(item.dependencies)}")
+            if item.notes:
+                lines.append(f"  - Notes: {item.notes}")
+                
+        return "\n".join(lines)
+    
+    @classmethod
+    def from_markdown(cls, markdown: str) -> 'TodoList':
+        """
+        Create TodoList from markdown format.
+        
+        Args:
+            markdown: Markdown string with checkbox format
+            
+        Returns:
+            TodoList instance
+        """
+        import re
+        
+        items = []
+        # Match checkbox items: - [ ] or - [x]
+        pattern = r'^-\s*\[([ x])\]\s*(.+?)$'
+        
+        for match in re.finditer(pattern, markdown, re.MULTILINE):
+            checked = match.group(1) == 'x'
+            description = match.group(2).strip()
+            
+            item = TodoItem(
+                description=description,
+                status="completed" if checked else "pending"
+            )
+            items.append(item)
+            
+        return cls(items=items)
+    
+    def to_json(self) -> str:
+        """
+        Serialize TodoList to JSON.
+        
+        Returns:
+            JSON string
+        """
+        data = {
+            "name": self.name,
+            "auto_update": self.auto_update,
+            "items": [item.to_dict() for item in self.items]
+        }
+        return json.dumps(data, indent=2)
+    
+    @classmethod
+    def from_json(cls, json_str: str) -> 'TodoList':
+        """
+        Create TodoList from JSON.
+        
+        Args:
+            json_str: JSON string
+            
+        Returns:
+            TodoList instance
+        """
+        data = json.loads(json_str)
+        items = [TodoItem.from_dict(item) for item in data.get("items", [])]
+        
+        return cls(
+            items=items,
+            auto_update=data.get("auto_update", True),
+            name=data.get("name", "Todo List")
+        )
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert TodoList to dictionary."""
+        return {
+            "name": self.name,
+            "auto_update": self.auto_update,
+            "items": [item.to_dict() for item in self.items]
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'TodoList':
+        """Create TodoList from dictionary."""
+        items = [TodoItem.from_dict(item) for item in data.get("items", [])]
+        return cls(
+            items=items,
+            auto_update=data.get("auto_update", True),
+            name=data.get("name", "Todo List")
+        )
+    
+    @classmethod
+    def from_plan(cls, plan: 'Plan') -> 'TodoList':
+        """
+        Create TodoList from a Plan.
+        
+        Args:
+            plan: Plan instance
+            
+        Returns:
+            TodoList with items matching plan steps
+        """
+        items = []
+        for step in plan.steps:
+            item = TodoItem(
+                id=step.id,
+                description=step.description,
+                status=step.status if step.status in ["pending", "in_progress", "completed"] else "pending",
+                dependencies=step.dependencies,
+                agent=step.agent
+            )
+            items.append(item)
+            
+        return cls(
+            items=items,
+            name=f"Todo: {plan.name}"
+        )
+    
+    def sync_with_plan(self, plan: 'Plan') -> None:
+        """
+        Synchronize TodoList with a Plan.
+        
+        Updates item statuses to match plan step statuses.
+        
+        Args:
+            plan: Plan to sync with
+        """
+        step_statuses = {step.id: step.status for step in plan.steps}
+        
+        for item in self.items:
+            if item.id in step_statuses:
+                status = step_statuses[item.id]
+                if status in ["pending", "in_progress", "completed"]:
+                    item.status = status
+                    
+    def update_from_plan(self, plan: 'Plan') -> None:
+        """
+        Update TodoList to match Plan structure.
+        
+        Adds new items for new steps, removes items for removed steps.
+        
+        Args:
+            plan: Plan to update from
+        """
+        # Get current step IDs
+        step_ids = {step.id for step in plan.steps}
+        item_ids = {item.id for item in self.items}
+        
+        # Remove items for deleted steps
+        self.items = [item for item in self.items if item.id in step_ids]
+        
+        # Add items for new steps
+        for step in plan.steps:
+            if step.id not in item_ids:
+                self.add(TodoItem(
+                    id=step.id,
+                    description=step.description,
+                    dependencies=step.dependencies,
+                    agent=step.agent
+                ))
+                
+        # Sync statuses
+        self.sync_with_plan(plan)

src/praisonai-agents/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "praisonaiagents"
-version = "0.0.180"
+version = "0.0.183"
 description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
 requires-python = ">=3.10"
 authors = [

src/praisonai-agents/test_planning_execution.py

@@ -0,0 +1,183 @@
+"""
+Complete Planning Mode Demo: Plan First, Then Execute Todo Items One by One.
+
+This demonstrates the full workflow:
+1. Create a plan with the PlanningAgent
+2. Generate todo list from the plan
+3. Execute each todo item sequentially with the appropriate agent
+4. Update progress as items are completed
+
+Run with: python test_planning_execution.py
+"""
+
+from praisonaiagents import Agent, Task
+from praisonaiagents.planning import PlanningAgent, TodoList
+
+
+def print_separator(title: str):
+    print("\n" + "=" * 70)
+    print(f"  {title}")
+    print("=" * 70 + "\n")
+
+
+def print_progress(todo_list: TodoList):
+    """Print current progress."""
+    completed = len(todo_list.completed)
+    total = len(todo_list.items)
+    progress = todo_list.progress * 100
+    bar_length = 30
+    filled = int(bar_length * todo_list.progress)
+    bar = "█" * filled + "░" * (bar_length - filled)
+    print(f"\n📊 Progress: [{bar}] {progress:.0f}% ({completed}/{total} completed)\n")
+
+
+def main():
+    print_separator("🎯 PLANNING MODE: PLAN FIRST, EXECUTE TODO ITEMS")
+    
+    # ==========================================================================
+    # STEP 1: Create Agents
+    # ==========================================================================
+    print("📦 Creating agents...")
+    
+    researcher = Agent(
+        name="Researcher",
+        role="Research Analyst",
+        goal="Research topics thoroughly and provide accurate information",
+        backstory="Expert researcher with access to vast knowledge bases",
+        verbose=True
+    )
+    
+    writer = Agent(
+        name="Writer",
+        role="Content Writer", 
+        goal="Write clear, engaging, and well-structured content",
+        backstory="Professional writer skilled in creating compelling narratives",
+        verbose=True
+    )
+    
+    print("✅ Agents created: Researcher, Writer\n")
+    
+    # ==========================================================================
+    # STEP 2: Create Plan using PlanningAgent
+    # ==========================================================================
+    print_separator("📋 PHASE 1: CREATING PLAN")
+    
+    planner = PlanningAgent(
+        llm="gpt-4o-mini",
+        read_only=True,
+        verbose=1
+    )
+    
+    request = "Write a short article about the top 3 benefits of meditation for mental health"
+    
+    print(f"📝 Request: {request}\n")
+    print("🤔 Planning agent is analyzing and creating a plan...\n")
+    
+    plan = planner.create_plan_sync(
+        request=request,
+        agents=[researcher, writer],
+        context="Keep it concise - each section should be 2-3 sentences max"
+    )
+    
+    print("\n✅ Plan created successfully!")
+    print(f"   Plan Name: {plan.name}")
+    print(f"   Total Steps: {len(plan.steps)}")
+    
+    # ==========================================================================
+    # STEP 3: Display the Plan
+    # ==========================================================================
+    print_separator("📄 GENERATED PLAN")
+    print(plan.to_markdown())
+    
+    # ==========================================================================
+    # STEP 4: Create Todo List from Plan
+    # ==========================================================================
+    print_separator("📝 PHASE 2: TODO LIST CREATED")
+    
+    todo_list = TodoList.from_plan(plan)
+    print(todo_list.to_markdown())
+    print_progress(todo_list)
+    
+    # ==========================================================================
+    # STEP 5: Execute Each Todo Item One by One
+    # ==========================================================================
+    print_separator("🚀 PHASE 3: EXECUTING TODO ITEMS")
+    
+    # Map agent names to agent instances
+    agent_map = {
+        "Researcher": researcher,
+        "Writer": writer
+    }
+    
+    results = {}
+    
+    for i, item in enumerate(todo_list.items):
+        print(f"\n{'─' * 70}")
+        print(f"📌 TODO ITEM {i + 1}/{len(todo_list.items)}")
+        print(f"{'─' * 70}")
+        print(f"   Description: {item.description}")
+        print(f"   Agent: {item.agent or 'Auto-assign'}")
+        print(f"   Status: {item.status}")
+        
+        if item.dependencies:
+            print(f"   Dependencies: {', '.join(item.dependencies)}")
+        
+        # Mark as in progress
+        todo_list.start(item.id)
+        print(f"\n   ⏳ Starting execution...")
+        
+        # Get the appropriate agent
+        agent_name = item.agent or "Researcher"
+        agent = agent_map.get(agent_name, researcher)
+        
+        # Build context from previous results
+        context = ""
+        for dep_id in item.dependencies:
+            if dep_id in results:
+                context += f"\nPrevious result ({dep_id}):\n{results[dep_id]}\n"
+        
+        # Create task description
+        task_description = item.description
+        if context:
+            task_description += f"\n\nContext from previous steps:{context}"
+        
+        # Execute with the agent
+        try:
+            result = agent.chat(task_description)
+            results[item.id] = result
+            
+            # Mark as completed
+            todo_list.complete(item.id)
+            
+            print(f"\n   ✅ Completed!")
+            print(f"\n   📄 Result Preview:")
+            print("   " + "-" * 50)
+            preview = str(result)[:500] + "..." if len(str(result)) > 500 else str(result)
+            for line in preview.split('\n'):
+                print(f"   {line}")
+            
+        except Exception as e:
+            print(f"\n   ❌ Error: {e}")
+            results[item.id] = f"Error: {e}"
+        
+        print_progress(todo_list)
+    
+    # ==========================================================================
+    # STEP 6: Final Summary
+    # ==========================================================================
+    print_separator("🎉 EXECUTION COMPLETE")
+    
+    print("📋 Final Todo List Status:")
+    print(todo_list.to_markdown())
+    
+    print("\n📊 Final Results Summary:")
+    print("-" * 50)
+    for item in todo_list.items:
+        status_icon = "✅" if item.status == "completed" else "❌"
+        print(f"{status_icon} {item.description[:60]}...")
+    
+    print(f"\n🏆 All {len(todo_list.completed)} tasks completed successfully!")
+
+
+if __name__ == "__main__":
+    main()

src/praisonai-agents/uv.lock

@@ -2422,7 +2422,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "0.0.180"
+version = "0.0.183"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },

src/praisonai/README.md

@@ -60,6 +60,7 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
 - 💾 **Prompt Caching** (Reduce costs & latency - OpenAI, Anthropic, Bedrock, Deepseek)
 - 🧠 **Claude Memory Tool** (Persistent cross-conversation memory - Anthropic Beta)
 - 💾 **File-Based Memory** (Zero-dependency persistent memory for all agents)
+- 📋 **Planning Mode** (Plan before execution like Cursor, Windsurf, Claude Code)
 
 ## Using Python Code
 

src/praisonai/praisonai.rb

@@ -3,8 +3,8 @@ class Praisonai < Formula
   
     desc "AI tools for various AI applications"
     homepage "https://github.com/MervinPraison/PraisonAI"
-    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.4.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.4.tar.gz | shasum -a 256`.split.first
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.5.tar.gz"
+    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.5.tar.gz | shasum -a 256`.split.first
     license "MIT"
   
     depends_on "[email protected]"

src/praisonai/praisonai/deploy.py

@@ -57,7 +57,7 @@ class CloudDeployer:
             file.write("FROM python:3.11-slim\n")
             file.write("WORKDIR /app\n")
             file.write("COPY . .\n")
-            file.write("RUN pip install flask praisonai==2.3.4 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==2.3.5 gunicorn markdown\n")
             file.write("EXPOSE 8080\n")
             file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
             

src/praisonai/praisonai/version.py

@@ -1 +1 @@
-__version__ = "2.3.4"
+__version__ = "2.3.5"

src/praisonai/pyproject.toml

@@ -12,7 +12,7 @@ dependencies = [
     "rich>=13.7",
     "markdown>=3.5",
     "pyparsing>=3.0.0",
-    "praisonaiagents>=0.0.180",
+    "praisonaiagents>=0.0.183",
     "python-dotenv>=0.19.0",
     "instructor>=1.3.3",
     "PyYAML>=6.0",

src/praisonai/uv.lock

@@ -4197,7 +4197,7 @@ requires-dist = [
     { name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
     { name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
     { name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
-    { name = "praisonaiagents", specifier = ">=0.0.180" },
+    { name = "praisonaiagents", specifier = ">=0.0.183" },
     { name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
     { name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
     { name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4252,7 +4252,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "0.0.180"
+version = "0.0.183"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "aiohttp" },
@@ -4261,9 +4261,9 @@ dependencies = [
     { name = "pydantic" },
     { name = "rich" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/a6/7e/8404716bff5d51dccada132bf2c9fb3986f858622f1f7651ff379d56a8fa/praisonaiagents-0.0.180.tar.gz", hash = "sha256:e668ac705e15b13478cd2b895c32cf603fbf49c380b8cfdb4949adefc8505014", size = 387604, upload-time = "2025-12-15T16:37:52.512Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/ba/fb/0971e0a371e5f8636ce34fb3fb556c8438db352f866a646df197998c7465/praisonaiagents-0.0.183.tar.gz", hash = "sha256:fccb8e97b43cda260ecd1f470eeb25abfa8c3f2610995bdede89a337054a4eab", size = 409014, upload-time = "2025-12-15T21:04:24.104Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/a0/ec/8e954809280beb9eee32c130fd32ee7f6c6e73f9d806228cb31b8bbc0752/praisonaiagents-0.0.180-py3-none-any.whl", hash = "sha256:c0d8dcde4deb4dd3029f1f1fe4b263a4e08783c803ad8253be1d11470a807157", size = 378348, upload-time = "2025-12-15T16:37:51.033Z" },
+    { url = "https://files.pythonhosted.org/packages/2a/e8/df5b543876604c2ea93e630dfb8482742a479c06d422b840f1f0d4430802/praisonaiagents-0.0.183-py3-none-any.whl", hash = "sha256:d5c4796dc74d639321e65aa09b9f15a74848a4a969c844260a7097801e76bda3", size = 397016, upload-time = "2025-12-15T21:04:21.085Z" },
 ]
 
 [[package]]