Release v2.3.14

## New Features:
- Context passing between workflow steps with {{previous_output}} and {{step_name_output}}
- Per-step agent configuration with role, goal, backstory, and tools
- Per-step tools support
- Async execution with aexecute() method
- Planning mode at workflow level
- Enhanced markdown parser for agent/tools blocks

## WorkflowStep Enhancements:
- agent_config: Optional[Dict] - Per-step agent configuration
- tools: Optional[List] - Per-step tools
- context_from: Optional[List[str]] - Specific steps to include context from
- retain_full_context: bool - Include all previous outputs vs only last
- output_variable: Optional[str] - Custom variable name for step output

## Workflow Enhancements:
- default_llm: Optional[str] - Default LLM for all steps
- planning: bool - Enable planning mode
- planning_llm: Optional[str] - LLM for planning
- memory_config: Optional[Dict] - Memory configuration

## Execute Method Enhancements:
- default_agent parameter (executor now optional)
- default_llm, memory, planning, stream, verbose parameters

## Tests:
- 74 new tests covering all features

README.md

@@ -40,7 +40,7 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
 | 🔄 Self Reflection AI Agents | [Example](examples/python/concepts/self-reflection-details.py) | [📖](https://docs.praison.ai/features/selfreflection) |
 | 🧠 Reasoning AI Agents | [Example](examples/python/concepts/reasoning-extraction.py) | [📖](https://docs.praison.ai/features/reasoning) |
 | 👁️ Multi Modal AI Agents | [Example](examples/python/general/multimodal.py) | [📖](https://docs.praison.ai/features/multimodal) |
-| 🎭 AI Agent Workflow | [Example](examples/python/stateful/workflow-state-example.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🎭 AI Agent Workflow (Context Passing, Per-Step Agents) | [Example](examples/python/stateful/workflow-state-example.py) | [📖](https://docs.praison.ai/features/workflows) |
 | 📚 Add Custom Knowledge | [Example](examples/python/concepts/knowledge-agents.py) | [📖](https://docs.praison.ai/features/knowledge) |
 | 🧠 Memory (Short & Long Term) | [Example](examples/python/general/memory_example.py) | [📖](https://docs.praison.ai/concepts/memory) |
 | 📄 Chat with PDF Agents | [Example](examples/python/concepts/chat-with-pdf.py) | [📖](https://docs.praison.ai/features/chat-with-pdf) |
@@ -392,21 +392,53 @@ memories = auto.process_interaction(
 
 ### 8. Workflows
 
-Create reusable multi-step workflows in `.praison/workflows/`:
+Create reusable multi-step workflows with context passing and per-step agents:
 
 ```python
-from praisonaiagents.memory import WorkflowManager
+from praisonaiagents import Agent
+from praisonaiagents.memory import WorkflowManager, Workflow, WorkflowStep
 
+# Simple execution with default agent
+agent = Agent(name="Assistant", llm="gpt-4o-mini")
 manager = WorkflowManager()
 
-# Execute a workflow
 result = manager.execute(
     "deploy",
-    executor=lambda prompt: agent.chat(prompt),
+    default_agent=agent,
     variables={"environment": "production"}
 )
+
+# Advanced: Per-step agent configuration
+workflow = Workflow(
+    name="research_pipeline",
+    default_llm="gpt-4o-mini",
+    steps=[
+        WorkflowStep(
+            name="research",
+            action="Research {{topic}}",
+            agent_config={"role": "Researcher", "goal": "Find information"},
+            tools=["tavily_search"]
+        ),
+        WorkflowStep(
+            name="write",
+            action="Write report based on {{previous_output}}",
+            agent_config={"role": "Writer", "goal": "Write content"},
+            context_from=["research"]  # Only include research output
+        )
+    ]
+)
+
+# Async execution
+import asyncio
+result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
 ```
 
+**Key Features:**
+- **Context Passing**: Use `{{previous_output}}` and `{{step_name_output}}` variables
+- **Per-Step Agents**: Configure different agents with roles, goals, tools for each step
+- **Async Execution**: Use `aexecute()` for async workflows
+- **Planning Mode**: Enable at workflow level with `planning=True`
+
 ### 9. Hooks
 
 Configure in `.praison/hooks.json`:

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.13" \
+    "praisonai>=2.3.14" \
     "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.13" \
+    "praisonai>=2.3.14" \
     "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.13" \
+    "praisonai>=2.3.14" \
     "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.13" \
+    "praisonai>=2.3.14" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

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

@@ -39,6 +39,15 @@ class WorkflowStep:
     on_error: Literal["stop", "continue", "retry"] = "stop"
     max_retries: int = 1
     
+    # Context passing fields
+    context_from: Optional[List[str]] = None  # Which previous steps to include context from
+    retain_full_context: bool = True  # Include all previous outputs vs only last
+    output_variable: Optional[str] = None  # Store output in this variable name
+    
+    # Agent configuration fields
+    agent_config: Optional[Dict[str, Any]] = None  # Per-step agent config {role, goal, backstory, llm}
+    tools: Optional[List[Any]] = None  # Tools for this step
+    
     def to_dict(self) -> Dict[str, Any]:
         return {
             "name": self.name,
@@ -46,7 +55,12 @@ class WorkflowStep:
             "action": self.action,
             "condition": self.condition,
             "on_error": self.on_error,
-            "max_retries": self.max_retries
+            "max_retries": self.max_retries,
+            "context_from": self.context_from,
+            "retain_full_context": self.retain_full_context,
+            "output_variable": self.output_variable,
+            "agent_config": self.agent_config,
+            "tools": self.tools
         }
 
 
@@ -59,13 +73,25 @@ class Workflow:
     variables: Dict[str, Any] = field(default_factory=dict)
     file_path: Optional[str] = None
     
+    # Default configuration for all steps
+    default_agent_config: Optional[Dict[str, Any]] = None  # Default agent for all steps
+    default_llm: Optional[str] = None  # Default LLM model
+    memory_config: Optional[Dict[str, Any]] = None  # Memory configuration
+    planning: bool = False  # Enable planning mode
+    planning_llm: Optional[str] = None  # LLM for planning
+    
     def to_dict(self) -> Dict[str, Any]:
         return {
             "name": self.name,
             "description": self.description,
             "steps": [s.to_dict() for s in self.steps],
             "variables": self.variables,
-            "file_path": self.file_path
+            "file_path": self.file_path,
+            "default_agent_config": self.default_agent_config,
+            "default_llm": self.default_llm,
+            "memory_config": self.memory_config,
+            "planning": self.planning,
+            "planning_llm": self.planning_llm
         }
 
 
@@ -198,6 +224,11 @@ class WorkflowManager:
         step_pattern = re.compile(r'^##\s+(?:Step\s+\d+[:\s]*)?(.+)$', re.MULTILINE)
         action_pattern = re.compile(r'```action\s*\n(.*?)\n```', re.DOTALL)
         condition_pattern = re.compile(r'```condition\s*\n(.*?)\n```', re.DOTALL)
+        agent_pattern = re.compile(r'```agent\s*\n(.*?)\n```', re.DOTALL)
+        tools_pattern = re.compile(r'```tools\s*\n(.*?)\n```', re.DOTALL)
+        context_from_pattern = re.compile(r'context_from:\s*\[([^\]]+)\]', re.IGNORECASE)
+        retain_context_pattern = re.compile(r'retain_full_context:\s*(true|false)', re.IGNORECASE)
+        output_var_pattern = re.compile(r'output_variable:\s*(\w+)', re.IGNORECASE)
         
         # Find all step headers
         matches = list(step_pattern.finditer(body))
@@ -218,23 +249,87 @@ class WorkflowManager:
             condition_match = condition_pattern.search(step_content)
             condition = condition_match.group(1).strip() if condition_match else None
             
+            # Extract agent config
+            agent_config = None
+            agent_match = agent_pattern.search(step_content)
+            if agent_match:
+                agent_config = self._parse_agent_config(agent_match.group(1).strip())
+            
+            # Extract tools list
+            tools = None
+            tools_match = tools_pattern.search(step_content)
+            if tools_match:
+                tools_str = tools_match.group(1).strip()
+                # Parse tools as list of strings (tool names)
+                tools = [t.strip().strip('"\'') for t in tools_str.split('\n') if t.strip()]
+            
+            # Extract context_from
+            context_from = None
+            context_from_match = context_from_pattern.search(step_content)
+            if context_from_match:
+                context_from = [s.strip().strip('"\'') for s in context_from_match.group(1).split(',')]
+            
+            # Extract retain_full_context
+            retain_full_context = True
+            retain_match = retain_context_pattern.search(step_content)
+            if retain_match:
+                retain_full_context = retain_match.group(1).lower() == 'true'
+            
+            # Extract output_variable
+            output_variable = None
+            output_var_match = output_var_pattern.search(step_content)
+            if output_var_match:
+                output_variable = output_var_match.group(1)
+            
             # Get description (text before action block)
             description = step_content
             if action_match:
                 description = step_content[:action_match.start()].strip()
             
-            # Clean up description (remove code blocks)
-            description = re.sub(r'```.*?```', '', description, flags=re.DOTALL).strip()
+            # Clean up description (remove code blocks and config lines)
+            description = re.sub(r'```.*?```', '', description, flags=re.DOTALL)
+            description = re.sub(r'context_from:.*', '', description)
+            description = re.sub(r'retain_full_context:.*', '', description)
+            description = re.sub(r'output_variable:.*', '', description)
+            description = description.strip()
             
             steps.append(WorkflowStep(
                 name=step_name,
                 description=description,
                 action=action,
-                condition=condition
+                condition=condition,
+                agent_config=agent_config,
+                tools=tools,
+                context_from=context_from,
+                retain_full_context=retain_full_context,
+                output_variable=output_variable
             ))
         
         return steps
     
+    def _parse_agent_config(self, agent_str: str) -> Dict[str, Any]:
+        """Parse agent configuration from a code block."""
+        config = {}
+        for line in agent_str.split('\n'):
+            line = line.strip()
+            if ':' in line and not line.startswith('#'):
+                key, value = line.split(':', 1)
+                key = key.strip()
+                value = value.strip()
+                
+                # Handle basic types
+                if value.lower() in ('true', 'false'):
+                    value = value.lower() == 'true'
+                elif value.isdigit():
+                    value = int(value)
+                elif value.startswith('"') and value.endswith('"'):
+                    value = value[1:-1]
+                elif value.startswith("'") and value.endswith("'"):
+                    value = value[1:-1]
+                
+                config[key] = value
+        return config
+    
     def _load_workflow(self, file_path: Path) -> Optional[Workflow]:
         """Load a workflow from a file."""
         try:
@@ -249,6 +344,13 @@ class WorkflowManager:
             if isinstance(variables, str):
                 variables = {}
             
+            # Parse workflow-level configuration
+            default_llm = frontmatter.get("default_llm")
+            planning = frontmatter.get("planning", False)
+            planning_llm = frontmatter.get("planning_llm")
+            memory_config = frontmatter.get("memory_config")
+            default_agent_config = frontmatter.get("default_agent_config")
+            
             steps = self._parse_steps(body)
             
             workflow = Workflow(
@@ -256,7 +358,12 @@ class WorkflowManager:
                 description=description,
                 steps=steps,
                 variables=variables,
-                file_path=str(file_path)
+                file_path=str(file_path),
+                default_llm=default_llm,
+                planning=planning,
+                planning_llm=planning_llm,
+                memory_config=memory_config,
+                default_agent_config=default_agent_config
             )
             
             self._log(f"Loaded workflow '{name}' with {len(steps)} steps")
@@ -310,23 +417,107 @@ class WorkflowManager:
         
         return re.sub(r'\{\{(\w+)\}\}', replace, text)
     
+    def _build_step_context(
+        self,
+        step: WorkflowStep,
+        step_index: int,
+        results: List[Dict[str, Any]],
+        variables: Dict[str, Any]
+    ) -> str:
+        """
+        Build context for a step from previous step outputs.
+        
+        Args:
+            step: Current workflow step
+            step_index: Index of current step
+            results: Results from previous steps
+            variables: Current variables dict
+            
+        Returns:
+            Context string to prepend to action
+        """
+        if step_index == 0 or not results:
+            return ""
+        
+        context_parts = []
+        
+        if step.context_from:
+            # Include only specified steps
+            for step_name in step.context_from:
+                for prev_result in results:
+                    if prev_result["step"] == step_name and prev_result.get("output"):
+                        context_parts.append(f"## {step_name} Output:\n{prev_result['output']}")
+        elif step.retain_full_context:
+            # Include all previous outputs
+            for prev_result in results:
+                if prev_result.get("output") and prev_result.get("status") == "success":
+                    context_parts.append(f"## {prev_result['step']} Output:\n{prev_result['output']}")
+        else:
+            # Include only the last step's output
+            last_result = results[-1]
+            if last_result.get("output") and last_result.get("status") == "success":
+                context_parts.append(f"## Previous Output:\n{last_result['output']}")
+        
+        if context_parts:
+            return "# Context from previous steps:\n\n" + "\n\n".join(context_parts) + "\n\n"
+        return ""
+    
+    def _update_variables_with_output(
+        self,
+        step: WorkflowStep,
+        output: str,
+        variables: Dict[str, Any],
+        results: List[Dict[str, Any]]
+    ) -> None:
+        """
+        Update variables dict with step output for variable substitution.
+        
+        Args:
+            step: Current workflow step
+            output: Step output
+            variables: Variables dict to update
+            results: Results list for previous_output
+        """
+        # Store output in custom variable name if specified
+        if step.output_variable:
+            variables[step.output_variable] = output
+        
+        # Store output with step name (normalized)
+        step_var_name = f"{step.name.lower().replace(' ', '_')}_output"
+        variables[step_var_name] = output
+        
+        # Update previous_output for next step
+        variables["previous_output"] = output
+    
     def execute(
         self,
         workflow_name: str,
-        executor: Callable[[str], str],
+        executor: Optional[Callable[[str], str]] = None,
         variables: Optional[Dict[str, Any]] = None,
         on_step: Optional[Callable[[WorkflowStep, int], None]] = None,
-        on_result: Optional[Callable[[WorkflowStep, str], None]] = None
+        on_result: Optional[Callable[[WorkflowStep, str], None]] = None,
+        default_agent: Optional[Any] = None,
+        default_llm: Optional[str] = None,
+        memory: Optional[Any] = None,
+        planning: bool = False,
+        stream: bool = False,
+        verbose: int = 0
     ) -> Dict[str, Any]:
         """
-        Execute a workflow.
+        Execute a workflow with context passing between steps.
         
         Args:
             workflow_name: Name of the workflow to execute
-            executor: Function to execute each step (e.g., agent.chat)
+            executor: Function to execute each step (e.g., agent.chat). Optional if default_agent provided.
             variables: Variables to substitute in steps
             on_step: Callback before each step (step, index)
             on_result: Callback after each step (step, result)
+            default_agent: Default agent to use for steps without agent_config
+            default_llm: Default LLM model for agent creation
+            memory: Shared memory instance
+            planning: Enable planning mode
+            stream: Enable streaming output
+            verbose: Verbosity level
             
         Returns:
             Execution results with step outputs and status
@@ -344,6 +535,12 @@ class WorkflowManager:
         if variables:
             all_variables.update(variables)
         
+        # Use workflow-level defaults if not provided
+        if default_llm is None:
+            default_llm = workflow.default_llm
+        if planning is False and workflow.planning:
+            planning = workflow.planning
+        
         results = []
         success = True
         
@@ -364,9 +561,39 @@ class WorkflowManager:
             if on_step:
                 on_step(step, i)
             
+            # Build context from previous steps
+            context = self._build_step_context(step, i, results, all_variables)
+            
             # Substitute variables in action
             action = self._substitute_variables(step.action, all_variables)
             
+            # Prepend context to action if available
+            if context:
+                action = f"{context}# Current Task:\n{action}"
+            
+            # Get or create executor for this step
+            step_executor = executor
+            if step_executor is None:
+                # Create agent for this step
+                step_agent = self._create_step_agent(
+                    step=step,
+                    default_agent=default_agent,
+                    default_llm=default_llm,
+                    memory=memory,
+                    verbose=verbose
+                )
+                if step_agent:
+                    def agent_executor(prompt, agent=step_agent):
+                        return agent.chat(prompt)
+                    step_executor = agent_executor
+            
+            if step_executor is None:
+                return {
+                    "success": False,
+                    "error": f"No executor available for step '{step.name}'. Provide executor or default_agent.",
+                    "results": results
+                }
+            
             # Execute step
             retries = 0
             step_success = False
@@ -375,7 +602,7 @@ class WorkflowManager:
             
             while retries <= step.max_retries and not step_success:
                 try:
-                    output = executor(action)
+                    output = step_executor(action)
                     step_success = True
                 except Exception as e:
                     error = str(e)
@@ -383,6 +610,10 @@ class WorkflowManager:
                     if retries <= step.max_retries:
                         self._log(f"Step '{step.name}' failed, retrying ({retries}/{step.max_retries})")
             
+            # Update variables with step output for next steps
+            if output and step_success:
+                self._update_variables_with_output(step, output, all_variables, results)
+            
             # Callback after step
             if on_result and output:
                 on_result(step, output)
@@ -405,9 +636,214 @@ class WorkflowManager:
         return {
             "success": success,
             "workflow": workflow.name,
-            "results": results
+            "results": results,
+            "variables": all_variables
         }
     
+    async def aexecute(
+        self,
+        workflow_name: str,
+        executor: Optional[Callable[[str], str]] = None,
+        variables: Optional[Dict[str, Any]] = None,
+        on_step: Optional[Callable[[WorkflowStep, int], None]] = None,
+        on_result: Optional[Callable[[WorkflowStep, str], None]] = None,
+        default_agent: Optional[Any] = None,
+        default_llm: Optional[str] = None,
+        memory: Optional[Any] = None,
+        planning: bool = False,
+        stream: bool = False,
+        verbose: int = 0
+    ) -> Dict[str, Any]:
+        """
+        Async version of execute() for workflow execution.
+        
+        Args:
+            workflow_name: Name of the workflow to execute
+            executor: Async function to execute each step. Optional if default_agent provided.
+            variables: Variables to substitute in steps
+            on_step: Callback before each step (step, index)
+            on_result: Callback after each step (step, result)
+            default_agent: Default agent to use for steps without agent_config
+            default_llm: Default LLM model for agent creation
+            memory: Shared memory instance
+            planning: Enable planning mode
+            stream: Enable streaming output
+            verbose: Verbosity level
+            
+        Returns:
+            Execution results with step outputs and status
+        """
+        import asyncio
+        
+        workflow = self.get_workflow(workflow_name)
+        if not workflow:
+            return {
+                "success": False,
+                "error": f"Workflow '{workflow_name}' not found",
+                "results": []
+            }
+        
+        # Merge variables
+        all_variables = {**workflow.variables}
+        if variables:
+            all_variables.update(variables)
+        
+        # Use workflow-level defaults if not provided
+        if default_llm is None:
+            default_llm = workflow.default_llm
+        if planning is False and workflow.planning:
+            planning = workflow.planning
+        
+        results = []
+        success = True
+        
+        for i, step in enumerate(workflow.steps):
+            # Check condition
+            if step.condition:
+                condition = self._substitute_variables(step.condition, all_variables)
+                if condition.lower() in ("false", "no", "skip", "0"):
+                    results.append({
+                        "step": step.name,
+                        "status": "skipped",
+                        "output": None
+                    })
+                    continue
+            
+            # Callback before step
+            if on_step:
+                on_step(step, i)
+            
+            # Build context from previous steps
+            context = self._build_step_context(step, i, results, all_variables)
+            
+            # Substitute variables in action
+            action = self._substitute_variables(step.action, all_variables)
+            
+            # Prepend context to action if available
+            if context:
+                action = f"{context}# Current Task:\n{action}"
+            
+            # Get or create executor for this step
+            step_executor = executor
+            if step_executor is None:
+                # Create agent for this step
+                step_agent = self._create_step_agent(
+                    step=step,
+                    default_agent=default_agent,
+                    default_llm=default_llm,
+                    memory=memory,
+                    verbose=verbose
+                )
+                if step_agent:
+                    # Check if agent has async chat method (and it's actually async)
+                    if hasattr(step_agent, 'achat') and asyncio.iscoroutinefunction(getattr(step_agent, 'achat', None)):
+                        async def async_agent_executor(prompt, agent=step_agent):
+                            return await agent.achat(prompt)
+                        step_executor = async_agent_executor
+                    else:
+                        def sync_agent_executor(prompt, agent=step_agent):
+                            return agent.chat(prompt)
+                        step_executor = sync_agent_executor
+            
+            if step_executor is None:
+                return {
+                    "success": False,
+                    "error": f"No executor available for step '{step.name}'. Provide executor or default_agent.",
+                    "results": results
+                }
+            
+            # Execute step
+            retries = 0
+            step_success = False
+            output = None
+            error = None
+            
+            while retries <= step.max_retries and not step_success:
+                try:
+                    # Check if executor is async
+                    if asyncio.iscoroutinefunction(step_executor):
+                        output = await step_executor(action)
+                    else:
+                        output = step_executor(action)
+                    step_success = True
+                except Exception as e:
+                    error = str(e)
+                    retries += 1
+                    if retries <= step.max_retries:
+                        self._log(f"Step '{step.name}' failed, retrying ({retries}/{step.max_retries})")
+            
+            # Update variables with step output for next steps
+            if output and step_success:
+                self._update_variables_with_output(step, output, all_variables, results)
+            
+            # Callback after step
+            if on_result and output:
+                on_result(step, output)
+            
+            results.append({
+                "step": step.name,
+                "status": "success" if step_success else "failed",
+                "output": output,
+                "error": error
+            })
+            
+            # Handle failure
+            if not step_success:
+                if step.on_error == "stop":
+                    success = False
+                    break
+                elif step.on_error == "continue":
+                    continue
+        
+        return {
+            "success": success,
+            "workflow": workflow.name,
+            "results": results,
+            "variables": all_variables
+        }
+    
+    def _create_step_agent(
+        self,
+        step: WorkflowStep,
+        default_agent: Optional[Any] = None,
+        default_llm: Optional[str] = None,
+        memory: Optional[Any] = None,
+        verbose: int = 0
+    ) -> Optional[Any]:
+        """
+        Create an agent for a specific step.
+        
+        Args:
+            step: Workflow step with optional agent_config
+            default_agent: Default agent to use if no config
+            default_llm: Default LLM model
+            memory: Shared memory instance
+            verbose: Verbosity level
+            
+        Returns:
+            Agent instance or None
+        """
+        if step.agent_config:
+            try:
+                from ..agent.agent import Agent
+                
+                config = step.agent_config.copy()
+                config.setdefault("name", f"{step.name}Agent")
+                config.setdefault("llm", default_llm)
+                config.setdefault("verbose", verbose)
+                
+                if step.tools:
+                    config["tools"] = step.tools
+                if memory:
+                    config["memory"] = memory
+                
+                return Agent(**config)
+            except Exception as e:
+                self._log(f"Failed to create agent for step '{step.name}': {e}", logging.ERROR)
+                return default_agent
+        
+        return default_agent
+    
     def create_workflow(
         self,
         name: str,

src/praisonai-agents/pyproject.toml

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

src/praisonai-agents/tests/test_workflow_agents.py

@@ -0,0 +1,390 @@
+"""
+Test suite for Workflow Per-Step Agent Configuration.
+
+Tests the enhanced workflow system's ability to configure different agents
+for each step, including custom roles, goals, backstories, and LLM settings.
+"""
+
+import pytest
+from unittest.mock import Mock, patch
+import sys
+import os
+
+# Add the package to path for testing
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from praisonaiagents.memory.workflows import (
+    WorkflowStep,
+    Workflow,
+    WorkflowManager
+)
+
+
+class TestWorkflowStepAgentFields:
+    """Test WorkflowStep dataclass agent-related fields."""
+    
+    def test_workflow_step_has_agent_config_field(self):
+        """WorkflowStep should have agent_config field."""
+        step = WorkflowStep(
+            name="test_step",
+            action="do something",
+            agent_config={
+                "role": "Researcher",
+                "goal": "Find information",
+                "backstory": "Expert researcher"
+            }
+        )
+        assert hasattr(step, 'agent_config')
+        assert step.agent_config["role"] == "Researcher"
+    
+    def test_workflow_step_has_tools_field(self):
+        """WorkflowStep should have tools field."""
+        step = WorkflowStep(
+            name="test_step",
+            action="do something",
+            tools=["search", "calculator"]
+        )
+        assert hasattr(step, 'tools')
+        assert "search" in step.tools
+    
+    def test_workflow_step_agent_config_default_none(self):
+        """agent_config should default to None."""
+        step = WorkflowStep(name="test_step", action="do something")
+        assert step.agent_config is None
+    
+    def test_workflow_step_tools_default_none(self):
+        """tools should default to None."""
+        step = WorkflowStep(name="test_step", action="do something")
+        assert step.tools is None
+    
+    def test_workflow_step_to_dict_includes_agent_fields(self):
+        """to_dict() should include agent-related fields."""
+        step = WorkflowStep(
+            name="test_step",
+            action="do something",
+            agent_config={"role": "Writer"},
+            tools=["grammar_check"]
+        )
+        d = step.to_dict()
+        assert "agent_config" in d
+        assert "tools" in d
+
+
+class TestWorkflowDefaultAgentConfig:
+    """Test Workflow dataclass default agent configuration."""
+    
+    def test_workflow_has_default_agent_config(self):
+        """Workflow should have default_agent_config field."""
+        workflow = Workflow(
+            name="test_workflow",
+            default_agent_config={
+                "role": "Assistant",
+                "goal": "Help with tasks"
+            }
+        )
+        assert hasattr(workflow, 'default_agent_config')
+        assert workflow.default_agent_config["role"] == "Assistant"
+    
+    def test_workflow_has_default_llm(self):
+        """Workflow should have default_llm field."""
+        workflow = Workflow(
+            name="test_workflow",
+            default_llm="gpt-4o-mini"
+        )
+        assert hasattr(workflow, 'default_llm')
+        assert workflow.default_llm == "gpt-4o-mini"
+    
+    def test_workflow_to_dict_includes_defaults(self):
+        """to_dict() should include default agent config fields."""
+        workflow = Workflow(
+            name="test_workflow",
+            default_agent_config={"role": "Helper"},
+            default_llm="gpt-4o"
+        )
+        d = workflow.to_dict()
+        assert "default_agent_config" in d
+        assert "default_llm" in d
+
+
+class TestCreateStepAgent:
+    """Test _create_step_agent method."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_create_step_agent_with_config(self):
+        """Should create agent from step's agent_config."""
+        manager = WorkflowManager()
+        
+        step = WorkflowStep(
+            name="research",
+            action="Research topic",
+            agent_config={
+                "role": "Researcher",
+                "goal": "Find comprehensive information",
+                "backstory": "Expert in research"
+            }
+        )
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            MockAgent.return_value = Mock()
+            agent = manager._create_step_agent(
+                step=step,
+                default_llm="gpt-4o-mini",
+                verbose=0
+            )
+            
+            # Verify Agent was called with config
+            MockAgent.assert_called_once()
+            call_kwargs = MockAgent.call_args[1]
+            assert call_kwargs["role"] == "Researcher"
+            assert call_kwargs["goal"] == "Find comprehensive information"
+    
+    def test_create_step_agent_uses_default_agent(self):
+        """Should return default_agent when no agent_config."""
+        manager = WorkflowManager()
+        
+        step = WorkflowStep(
+            name="simple_step",
+            action="Do something"
+        )
+        
+        default_agent = Mock()
+        agent = manager._create_step_agent(
+            step=step,
+            default_agent=default_agent,
+            default_llm="gpt-4o-mini"
+        )
+        
+        assert agent == default_agent
+    
+    def test_create_step_agent_includes_tools(self):
+        """Should include tools in agent config."""
+        manager = WorkflowManager()
+        
+        mock_tool = Mock()
+        step = WorkflowStep(
+            name="research",
+            action="Research topic",
+            agent_config={"role": "Researcher"},
+            tools=[mock_tool]
+        )
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            MockAgent.return_value = Mock()
+            manager._create_step_agent(
+                step=step,
+                default_llm="gpt-4o-mini"
+            )
+            
+            call_kwargs = MockAgent.call_args[1]
+            assert "tools" in call_kwargs
+            assert mock_tool in call_kwargs["tools"]
+    
+    def test_create_step_agent_includes_memory(self):
+        """Should include memory in agent config."""
+        manager = WorkflowManager()
+        
+        step = WorkflowStep(
+            name="research",
+            action="Research topic",
+            agent_config={"role": "Researcher"}
+        )
+        
+        mock_memory = Mock()
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            MockAgent.return_value = Mock()
+            manager._create_step_agent(
+                step=step,
+                memory=mock_memory
+            )
+            
+            call_kwargs = MockAgent.call_args[1]
+            assert "memory" in call_kwargs
+            assert call_kwargs["memory"] == mock_memory
+
+
+class TestExecuteWithAgents:
+    """Test execute() with per-step agent configuration."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_execute_with_default_agent(self):
+        """Execute should use default_agent when provided."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1"),
+                WorkflowStep(name="step2", action="Action 2")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        # Create mock default agent
+        mock_agent = Mock()
+        mock_agent.chat = Mock(return_value="Agent response")
+        
+        result = manager.execute(
+            "test_workflow",
+            default_agent=mock_agent
+        )
+        
+        assert result["success"]
+        assert mock_agent.chat.call_count == 2
+    
+    def test_execute_with_step_agent_config(self):
+        """Execute should create agent from step's agent_config."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(
+                    name="research",
+                    action="Research AI",
+                    agent_config={
+                        "role": "Researcher",
+                        "goal": "Find information"
+                    }
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            mock_agent_instance = Mock()
+            mock_agent_instance.chat = Mock(return_value="Research results")
+            MockAgent.return_value = mock_agent_instance
+            
+            result = manager.execute(
+                "test_workflow",
+                default_llm="gpt-4o-mini"
+            )
+            
+            assert result["success"]
+            MockAgent.assert_called_once()
+    
+    def test_execute_with_executor_overrides_agent(self):
+        """Executor function should override agent creation."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(
+                    name="step1",
+                    action="Action 1",
+                    agent_config={"role": "Should not be used"}
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        executor_called = [False]
+        
+        def custom_executor(action):
+            executor_called[0] = True
+            return "Custom executor result"
+        
+        result = manager.execute(
+            "test_workflow",
+            executor=custom_executor
+        )
+        
+        assert result["success"]
+        assert executor_called[0]
+    
+    def test_execute_fails_without_executor_or_agent(self):
+        """Execute should fail if no executor or agent provided."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        result = manager.execute("test_workflow")
+        
+        assert not result["success"]
+        assert "No executor available" in result.get("error", "")
+
+
+class TestWorkflowLevelDefaults:
+    """Test workflow-level default configurations."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_workflow_default_llm_used(self):
+        """Workflow's default_llm should be used for agent creation."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            default_llm="gpt-4o",
+            steps=[
+                WorkflowStep(
+                    name="step1",
+                    action="Action 1",
+                    agent_config={"role": "Helper"}
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            mock_agent_instance = Mock()
+            mock_agent_instance.chat = Mock(return_value="Response")
+            MockAgent.return_value = mock_agent_instance
+            
+            manager.execute("test_workflow")
+            
+            call_kwargs = MockAgent.call_args[1]
+            assert call_kwargs.get("llm") == "gpt-4o"
+    
+    def test_execute_default_llm_overrides_workflow(self):
+        """Execute's default_llm should override workflow's default_llm."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            default_llm="gpt-3.5-turbo",
+            steps=[
+                WorkflowStep(
+                    name="step1",
+                    action="Action 1",
+                    agent_config={"role": "Helper"}
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            mock_agent_instance = Mock()
+            mock_agent_instance.chat = Mock(return_value="Response")
+            MockAgent.return_value = mock_agent_instance
+            
+            manager.execute("test_workflow", default_llm="gpt-4o-mini")
+            
+            call_kwargs = MockAgent.call_args[1]
+            assert call_kwargs.get("llm") == "gpt-4o-mini"
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

src/praisonai-agents/tests/test_workflow_async.py

@@ -0,0 +1,202 @@
+"""
+Test suite for Workflow Async Support.
+
+Tests the aexecute() async method for workflow execution.
+"""
+
+import pytest
+import asyncio
+from unittest.mock import Mock, AsyncMock, patch
+import sys
+import os
+
+# Add the package to path for testing
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from praisonaiagents.memory.workflows import (
+    WorkflowStep,
+    Workflow,
+    WorkflowManager
+)
+
+
+class TestAexecuteMethod:
+    """Test aexecute() async method."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    @pytest.mark.asyncio
+    async def test_aexecute_exists(self):
+        """WorkflowManager should have aexecute method."""
+        manager = WorkflowManager()
+        assert hasattr(manager, 'aexecute')
+        assert asyncio.iscoroutinefunction(manager.aexecute)
+    
+    @pytest.mark.asyncio
+    async def test_aexecute_with_sync_executor(self):
+        """aexecute should work with sync executor."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1"),
+                WorkflowStep(name="step2", action="Action 2")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        def sync_executor(action):
+            return f"Sync result for {action}"
+        
+        result = await manager.aexecute("test_workflow", executor=sync_executor)
+        
+        assert result["success"]
+        assert len(result["results"]) == 2
+    
+    @pytest.mark.asyncio
+    async def test_aexecute_with_async_executor(self):
+        """aexecute should work with async executor."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1"),
+                WorkflowStep(name="step2", action="Action 2")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        async def async_executor(action):
+            await asyncio.sleep(0.01)  # Simulate async operation
+            return f"Async result for {action}"
+        
+        result = await manager.aexecute("test_workflow", executor=async_executor)
+        
+        assert result["success"]
+        assert len(result["results"]) == 2
+        assert "Async result" in result["results"][0]["output"]
+    
+    @pytest.mark.asyncio
+    async def test_aexecute_context_passing(self):
+        """aexecute should pass context between steps."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Generate data"),
+                WorkflowStep(name="step2", action="Process {{previous_output}}")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        received_actions = []
+        
+        async def async_executor(action):
+            received_actions.append(action)
+            if "Generate" in action:
+                return "DATA_ABC"
+            return "Processed"
+        
+        result = await manager.aexecute("test_workflow", executor=async_executor)
+        
+        assert result["success"]
+        # Step 2 should have received context
+        assert "DATA_ABC" in received_actions[1] or "previous_output" in received_actions[1]
+    
+    @pytest.mark.asyncio
+    async def test_aexecute_workflow_not_found(self):
+        """aexecute should return error for non-existent workflow."""
+        manager = WorkflowManager()
+        manager._loaded = True
+        
+        result = await manager.aexecute("nonexistent_workflow")
+        
+        assert not result["success"]
+        assert "not found" in result.get("error", "")
+    
+    @pytest.mark.asyncio
+    async def test_aexecute_with_default_agent(self):
+        """aexecute should work with default_agent."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        mock_agent = Mock()
+        mock_agent.chat = Mock(return_value="Agent response")
+        
+        result = await manager.aexecute(
+            "test_workflow",
+            default_agent=mock_agent
+        )
+        
+        assert result["success"]
+        mock_agent.chat.assert_called()
+    
+    @pytest.mark.asyncio
+    async def test_aexecute_with_async_agent(self):
+        """aexecute should use achat if available on agent."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(
+                    name="step1",
+                    action="Action 1",
+                    agent_config={"role": "Helper"}
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            mock_agent = Mock()
+            mock_agent.achat = AsyncMock(return_value="Async agent response")
+            MockAgent.return_value = mock_agent
+            
+            result = await manager.aexecute(
+                "test_workflow",
+                default_llm="gpt-4o-mini"
+            )
+            
+            assert result["success"]
+            mock_agent.achat.assert_called()
+    
+    @pytest.mark.asyncio
+    async def test_aexecute_returns_variables(self):
+        """aexecute should return accumulated variables."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            variables={"initial": "value"},
+            steps=[
+                WorkflowStep(name="step1", action="Action 1")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        async def async_executor(action):
+            return "Output"
+        
+        result = await manager.aexecute("test_workflow", executor=async_executor)
+        
+        assert "variables" in result
+        assert "previous_output" in result["variables"]
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

src/praisonai-agents/tests/test_workflow_context_passing.py

@@ -0,0 +1,353 @@
+"""
+Test suite for Workflow Context Passing functionality.
+
+Tests the enhanced workflow system's ability to pass context between steps,
+including output accumulation, variable substitution, and selective context inclusion.
+"""
+
+import pytest
+from unittest.mock import Mock, MagicMock, patch
+import sys
+import os
+
+# Add the package to path for testing
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from praisonaiagents.memory.workflows import (
+    WorkflowStep,
+    Workflow,
+    WorkflowManager,
+    create_workflow_manager
+)
+
+
+class TestWorkflowStepContextFields:
+    """Test WorkflowStep dataclass context-related fields."""
+    
+    def test_workflow_step_has_context_from_field(self):
+        """WorkflowStep should have context_from field."""
+        step = WorkflowStep(
+            name="test_step",
+            action="do something",
+            context_from=["step1", "step2"]
+        )
+        assert hasattr(step, 'context_from')
+        assert step.context_from == ["step1", "step2"]
+    
+    def test_workflow_step_has_retain_full_context_field(self):
+        """WorkflowStep should have retain_full_context field with default True."""
+        step = WorkflowStep(name="test_step", action="do something")
+        assert hasattr(step, 'retain_full_context')
+        assert step.retain_full_context == True
+    
+    def test_workflow_step_has_output_variable_field(self):
+        """WorkflowStep should have output_variable field."""
+        step = WorkflowStep(
+            name="test_step",
+            action="do something",
+            output_variable="research_result"
+        )
+        assert hasattr(step, 'output_variable')
+        assert step.output_variable == "research_result"
+    
+    def test_workflow_step_to_dict_includes_new_fields(self):
+        """to_dict() should include context-related fields."""
+        step = WorkflowStep(
+            name="test_step",
+            action="do something",
+            context_from=["step1"],
+            retain_full_context=False,
+            output_variable="result"
+        )
+        d = step.to_dict()
+        assert "context_from" in d
+        assert "retain_full_context" in d
+        assert "output_variable" in d
+
+
+class TestContextPassing:
+    """Test context passing between workflow steps."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_step_receives_previous_step_output(self):
+        """Step 2 should receive Step 1's output as context."""
+        manager = WorkflowManager()
+        
+        # Create a simple workflow with 2 steps
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Generate data"),
+                WorkflowStep(name="step2", action="Process {{previous_output}}")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        # Track what each step receives
+        received_actions = []
+        
+        def mock_executor(action):
+            received_actions.append(action)
+            if "Generate" in action:
+                return "Generated data: ABC123"
+            return f"Processed: {action}"
+        
+        result = manager.execute("test_workflow", executor=mock_executor)
+        
+        assert result["success"] == True
+        assert len(result["results"]) == 2
+        # Step 2 should have received the output from step 1
+        assert "ABC123" in received_actions[1] or "previous_output" in received_actions[1]
+    
+    def test_context_accumulates_across_steps(self):
+        """Each step should have access to all previous outputs when retain_full_context=True."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="First action", retain_full_context=True),
+                WorkflowStep(name="step2", action="Second action", retain_full_context=True),
+                WorkflowStep(name="step3", action="Third with context: {{step1_output}} and {{step2_output}}", retain_full_context=True)
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        outputs = ["Output1", "Output2", "Output3"]
+        call_count = [0]
+        
+        def mock_executor(action):
+            idx = call_count[0]
+            call_count[0] += 1
+            return outputs[idx]
+        
+        result = manager.execute("test_workflow", executor=mock_executor)
+        
+        assert result["success"] == True
+        assert len(result["results"]) == 3
+        # All outputs should be stored
+        assert result["results"][0]["output"] == "Output1"
+        assert result["results"][1]["output"] == "Output2"
+        assert result["results"][2]["output"] == "Output3"
+    
+    def test_context_from_specific_steps(self):
+        """Step should only receive context from specified steps via context_from."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="research", action="Research topic"),
+                WorkflowStep(name="analysis", action="Analyze data"),
+                WorkflowStep(
+                    name="summary",
+                    action="Summarize based on: {{research_output}}",
+                    context_from=["research"],  # Only include research, not analysis
+                    retain_full_context=False
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        received_actions = []
+        
+        def mock_executor(action):
+            received_actions.append(action)
+            if "Research" in action:
+                return "Research findings: XYZ"
+            elif "Analyze" in action:
+                return "Analysis results: 123"
+            return "Summary complete"
+        
+        result = manager.execute("test_workflow", executor=mock_executor)
+        
+        assert result["success"] == True
+        # The summary step should have research output but not analysis
+        # (This tests the context_from filtering)
+    
+    def test_output_variable_substitution(self):
+        """Output should be stored in specified variable name."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(
+                    name="step1",
+                    action="Generate report",
+                    output_variable="report_data"
+                ),
+                WorkflowStep(
+                    name="step2",
+                    action="Use report: {{report_data}}"
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        def mock_executor(action):
+            if "Generate" in action:
+                return "Report content here"
+            return f"Used: {action}"
+        
+        result = manager.execute("test_workflow", executor=mock_executor)
+        
+        assert result["success"] == True
+    
+    def test_retain_full_context_false_only_last_step(self):
+        """When retain_full_context=False, only include last step's output."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="First"),
+                WorkflowStep(name="step2", action="Second"),
+                WorkflowStep(
+                    name="step3",
+                    action="Third with {{previous_output}}",
+                    retain_full_context=False
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        received_actions = []
+        
+        def mock_executor(action):
+            received_actions.append(action)
+            if "First" in action:
+                return "First output"
+            elif "Second" in action:
+                return "Second output"
+            return "Third output"
+        
+        result = manager.execute("test_workflow", executor=mock_executor)
+        
+        assert result["success"] == True
+        # Step 3 should only have step 2's output, not step 1's
+
+
+class TestContextVariableSubstitution:
+    """Test variable substitution with step outputs."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_previous_output_variable(self):
+        """{{previous_output}} should be replaced with last step's output."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Generate"),
+                WorkflowStep(name="step2", action="Process: {{previous_output}}")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        received_actions = []
+        
+        def mock_executor(action):
+            received_actions.append(action)
+            if "Generate" in action:
+                return "DATA_123"
+            return "Done"
+        
+        result = manager.execute("test_workflow", executor=mock_executor)
+        
+        # Check that previous_output was substituted
+        assert "DATA_123" in received_actions[1] or "previous_output" in received_actions[1]
+    
+    def test_named_step_output_variable(self):
+        """{{step_name_output}} should be replaced with that step's output."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="research", action="Research topic"),
+                WorkflowStep(name="write", action="Write about {{research_output}}")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        received_actions = []
+        
+        def mock_executor(action):
+            received_actions.append(action)
+            if "Research" in action:
+                return "Research findings"
+            return "Article written"
+        
+        result = manager.execute("test_workflow", executor=mock_executor)
+        
+        assert result["success"] == True
+
+
+class TestWorkflowExecuteResults:
+    """Test that execute() returns proper results with context info."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_results_include_step_outputs(self):
+        """Results should include output from each step."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1"),
+                WorkflowStep(name="step2", action="Action 2")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        def mock_executor(action):
+            return f"Output for {action}"
+        
+        result = manager.execute("test_workflow", executor=mock_executor)
+        
+        assert result["success"] == True
+        assert len(result["results"]) == 2
+        assert result["results"][0]["output"] is not None
+        assert result["results"][1]["output"] is not None
+    
+    def test_results_include_accumulated_context(self):
+        """Results should include accumulated context information."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1"),
+                WorkflowStep(name="step2", action="Action 2")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        def mock_executor(action):
+            return f"Output for {action}"
+        
+        result = manager.execute("test_workflow", executor=mock_executor)
+        
+        # Result should have context or variables with accumulated outputs
+        assert "results" in result
+        assert result["success"]
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

src/praisonai-agents/tests/test_workflow_markdown_parser.py

@@ -0,0 +1,341 @@
+"""
+Test suite for Workflow Markdown Parser.
+
+Tests the enhanced markdown parser for parsing agent config, tools,
+context_from, and other new fields from workflow markdown files.
+"""
+
+import pytest
+import tempfile
+import os
+import sys
+
+# Add the package to path for testing
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from praisonaiagents.memory.workflows import (
+    WorkflowStep,
+    Workflow,
+    WorkflowManager
+)
+
+
+class TestParseAgentConfig:
+    """Test parsing agent configuration from markdown."""
+    
+    def test_parse_agent_config_block(self):
+        """Should parse agent config from ```agent block."""
+        manager = WorkflowManager()
+        
+        body = '''
+## Step 1: Research
+Research the topic.
+
+```agent
+role: Researcher
+goal: Find comprehensive information
+backstory: Expert researcher with 10 years experience
+llm: gpt-4o
+```
+
+```action
+Search for information about AI
+```
+'''
+        steps = manager._parse_steps(body)
+        
+        assert len(steps) == 1
+        assert steps[0].agent_config is not None
+        assert steps[0].agent_config["role"] == "Researcher"
+        assert steps[0].agent_config["goal"] == "Find comprehensive information"
+        assert steps[0].agent_config["llm"] == "gpt-4o"
+    
+    def test_parse_agent_config_helper(self):
+        """_parse_agent_config should parse key-value pairs."""
+        manager = WorkflowManager()
+        
+        agent_str = '''
+role: Writer
+goal: Write engaging content
+verbose: true
+max_iter: 10
+'''
+        config = manager._parse_agent_config(agent_str)
+        
+        assert config["role"] == "Writer"
+        assert config["goal"] == "Write engaging content"
+        assert config["verbose"] is True
+        assert config["max_iter"] == 10
+
+
+class TestParseToolsBlock:
+    """Test parsing tools from markdown."""
+    
+    def test_parse_tools_block(self):
+        """Should parse tools from ```tools block."""
+        manager = WorkflowManager()
+        
+        body = '''
+## Step 1: Search
+Search for information.
+
+```tools
+tavily_search
+web_browser
+calculator
+```
+
+```action
+Search for AI trends
+```
+'''
+        steps = manager._parse_steps(body)
+        
+        assert len(steps) == 1
+        assert steps[0].tools is not None
+        assert len(steps[0].tools) == 3
+        assert "tavily_search" in steps[0].tools
+        assert "web_browser" in steps[0].tools
+        assert "calculator" in steps[0].tools
+
+
+class TestParseContextFrom:
+    """Test parsing context_from from markdown."""
+    
+    def test_parse_context_from(self):
+        """Should parse context_from directive."""
+        manager = WorkflowManager()
+        
+        body = '''
+## Step 1: Research
+Research the topic.
+
+```action
+Research AI
+```
+
+## Step 2: Analyze
+Analyze the research.
+
+```action
+Analyze data
+```
+
+## Step 3: Summary
+Summarize findings.
+
+context_from: [Research, Analyze]
+
+```action
+Write summary
+```
+'''
+        steps = manager._parse_steps(body)
+        
+        assert len(steps) == 3
+        assert steps[2].context_from is not None
+        assert "Research" in steps[2].context_from
+        assert "Analyze" in steps[2].context_from
+
+
+class TestParseRetainFullContext:
+    """Test parsing retain_full_context from markdown."""
+    
+    def test_parse_retain_full_context_false(self):
+        """Should parse retain_full_context: false."""
+        manager = WorkflowManager()
+        
+        body = '''
+## Step 1: Process
+Process data.
+
+retain_full_context: false
+
+```action
+Process the data
+```
+'''
+        steps = manager._parse_steps(body)
+        
+        assert len(steps) == 1
+        assert steps[0].retain_full_context is False
+    
+    def test_parse_retain_full_context_true(self):
+        """Should parse retain_full_context: true."""
+        manager = WorkflowManager()
+        
+        body = '''
+## Step 1: Process
+Process data.
+
+retain_full_context: true
+
+```action
+Process the data
+```
+'''
+        steps = manager._parse_steps(body)
+        
+        assert len(steps) == 1
+        assert steps[0].retain_full_context is True
+    
+    def test_retain_full_context_default_true(self):
+        """retain_full_context should default to True."""
+        manager = WorkflowManager()
+        
+        body = '''
+## Step 1: Process
+Process data.
+
+```action
+Process the data
+```
+'''
+        steps = manager._parse_steps(body)
+        
+        assert len(steps) == 1
+        assert steps[0].retain_full_context is True
+
+
+class TestParseOutputVariable:
+    """Test parsing output_variable from markdown."""
+    
+    def test_parse_output_variable(self):
+        """Should parse output_variable directive."""
+        manager = WorkflowManager()
+        
+        body = '''
+## Step 1: Generate
+Generate report.
+
+output_variable: report_data
+
+```action
+Generate the report
+```
+'''
+        steps = manager._parse_steps(body)
+        
+        assert len(steps) == 1
+        assert steps[0].output_variable == "report_data"
+
+
+class TestLoadWorkflowWithNewFields:
+    """Test loading workflow files with new fields."""
+    
+    def test_load_workflow_with_default_llm(self):
+        """Should load workflow with default_llm from frontmatter."""
+        manager = WorkflowManager()
+        
+        content = '''---
+name: Test Workflow
+description: A test workflow
+default_llm: gpt-4o
+planning: true
+planning_llm: gpt-4o-mini
+---
+
+## Step 1: Test
+Test step.
+
+```action
+Do something
+```
+'''
+        # Create temp file
+        with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False) as f:
+            f.write(content)
+            temp_path = f.name
+        
+        try:
+            from pathlib import Path
+            workflow = manager._load_workflow(Path(temp_path))
+            
+            assert workflow is not None
+            assert workflow.default_llm == "gpt-4o"
+            assert workflow.planning is True
+            assert workflow.planning_llm == "gpt-4o-mini"
+        finally:
+            os.unlink(temp_path)
+
+
+class TestCompleteWorkflowParsing:
+    """Test complete workflow parsing with all new features."""
+    
+    def test_parse_complete_workflow(self):
+        """Should parse a complete workflow with all new features."""
+        manager = WorkflowManager()
+        
+        body = '''
+## Step 1: Research
+Research the topic thoroughly.
+
+```agent
+role: Researcher
+goal: Find comprehensive information
+backstory: Expert researcher
+```
+
+```tools
+tavily_search
+web_browser
+```
+
+output_variable: research_data
+
+```action
+Search for information about {{topic}}
+```
+
+## Step 2: Analyze
+Analyze the research findings.
+
+```agent
+role: Analyst
+goal: Analyze data patterns
+```
+
+context_from: [Research]
+retain_full_context: false
+
+```action
+Analyze: {{research_data}}
+```
+
+## Step 3: Write
+Write the final report.
+
+```agent
+role: Writer
+goal: Write engaging content
+```
+
+```action
+Write report based on analysis
+```
+'''
+        steps = manager._parse_steps(body)
+        
+        assert len(steps) == 3
+        
+        # Step 1
+        assert steps[0].name == "Research"
+        assert steps[0].agent_config is not None
+        assert steps[0].agent_config["role"] == "Researcher"
+        assert steps[0].tools is not None
+        assert "tavily_search" in steps[0].tools
+        assert steps[0].output_variable == "research_data"
+        
+        # Step 2
+        assert steps[1].name == "Analyze"
+        assert steps[1].agent_config["role"] == "Analyst"
+        assert steps[1].context_from == ["Research"]
+        assert steps[1].retain_full_context is False
+        
+        # Step 3
+        assert steps[2].name == "Write"
+        assert steps[2].agent_config["role"] == "Writer"
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

src/praisonai-agents/tests/test_workflow_planning.py

@@ -0,0 +1,359 @@
+"""
+Test suite for Workflow Planning Mode Integration.
+
+Tests the enhanced workflow system's ability to integrate with planning mode,
+including workflow-level planning settings and planning LLM configuration.
+"""
+
+import pytest
+from unittest.mock import Mock, patch
+import sys
+import os
+
+# Add the package to path for testing
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from praisonaiagents.memory.workflows import (
+    WorkflowStep,
+    Workflow,
+    WorkflowManager
+)
+
+
+class TestWorkflowPlanningFields:
+    """Test Workflow dataclass planning-related fields."""
+    
+    def test_workflow_has_planning_field(self):
+        """Workflow should have planning field."""
+        workflow = Workflow(
+            name="test_workflow",
+            planning=True
+        )
+        assert hasattr(workflow, 'planning')
+        assert workflow.planning is True
+    
+    def test_workflow_planning_default_false(self):
+        """Workflow planning should default to False."""
+        workflow = Workflow(name="test_workflow")
+        assert workflow.planning is False
+    
+    def test_workflow_has_planning_llm_field(self):
+        """Workflow should have planning_llm field."""
+        workflow = Workflow(
+            name="test_workflow",
+            planning_llm="gpt-4o"
+        )
+        assert hasattr(workflow, 'planning_llm')
+        assert workflow.planning_llm == "gpt-4o"
+    
+    def test_workflow_planning_llm_default_none(self):
+        """Workflow planning_llm should default to None."""
+        workflow = Workflow(name="test_workflow")
+        assert workflow.planning_llm is None
+    
+    def test_workflow_to_dict_includes_planning_fields(self):
+        """to_dict() should include planning-related fields."""
+        workflow = Workflow(
+            name="test_workflow",
+            planning=True,
+            planning_llm="gpt-4o"
+        )
+        d = workflow.to_dict()
+        assert "planning" in d
+        assert "planning_llm" in d
+        assert d["planning"] is True
+        assert d["planning_llm"] == "gpt-4o"
+
+
+class TestWorkflowMemoryFields:
+    """Test Workflow dataclass memory-related fields."""
+    
+    def test_workflow_has_memory_config_field(self):
+        """Workflow should have memory_config field."""
+        workflow = Workflow(
+            name="test_workflow",
+            memory_config={
+                "provider": "rag",
+                "use_embedding": True
+            }
+        )
+        assert hasattr(workflow, 'memory_config')
+        assert workflow.memory_config["provider"] == "rag"
+    
+    def test_workflow_memory_config_default_none(self):
+        """Workflow memory_config should default to None."""
+        workflow = Workflow(name="test_workflow")
+        assert workflow.memory_config is None
+    
+    def test_workflow_to_dict_includes_memory_config(self):
+        """to_dict() should include memory_config field."""
+        workflow = Workflow(
+            name="test_workflow",
+            memory_config={"provider": "sqlite"}
+        )
+        d = workflow.to_dict()
+        assert "memory_config" in d
+
+
+class TestExecutePlanningParameter:
+    """Test execute() planning parameter."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_execute_accepts_planning_parameter(self):
+        """Execute should accept planning parameter."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        def mock_executor(action):
+            return "Result"
+        
+        # Should not raise error
+        result = manager.execute(
+            "test_workflow",
+            executor=mock_executor,
+            planning=True
+        )
+        
+        assert result["success"]
+    
+    def test_workflow_planning_used_when_not_overridden(self):
+        """Workflow's planning setting should be used when not overridden."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            planning=True,  # Workflow-level planning enabled
+            steps=[
+                WorkflowStep(name="step1", action="Action 1")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        def mock_executor(action):
+            return "Result"
+        
+        # Execute without planning parameter - should use workflow's setting
+        result = manager.execute(
+            "test_workflow",
+            executor=mock_executor
+        )
+        
+        assert result["success"]
+
+
+class TestExecuteMemoryParameter:
+    """Test execute() memory parameter."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_execute_accepts_memory_parameter(self):
+        """Execute should accept memory parameter."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        mock_memory = Mock()
+        
+        def mock_executor(action):
+            return "Result"
+        
+        # Should not raise error
+        result = manager.execute(
+            "test_workflow",
+            executor=mock_executor,
+            memory=mock_memory
+        )
+        
+        assert result["success"]
+    
+    def test_memory_passed_to_step_agent(self):
+        """Memory should be passed to created step agents."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(
+                    name="step1",
+                    action="Action 1",
+                    agent_config={"role": "Helper"}
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        mock_memory = Mock()
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            mock_agent = Mock()
+            mock_agent.chat = Mock(return_value="Response")
+            MockAgent.return_value = mock_agent
+            
+            manager.execute(
+                "test_workflow",
+                memory=mock_memory,
+                default_llm="gpt-4o-mini"
+            )
+            
+            call_kwargs = MockAgent.call_args[1]
+            assert "memory" in call_kwargs
+            assert call_kwargs["memory"] == mock_memory
+
+
+class TestExecuteStreamParameter:
+    """Test execute() stream parameter."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_execute_accepts_stream_parameter(self):
+        """Execute should accept stream parameter."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        def mock_executor(action):
+            return "Result"
+        
+        # Should not raise error
+        result = manager.execute(
+            "test_workflow",
+            executor=mock_executor,
+            stream=True
+        )
+        
+        assert result["success"]
+
+
+class TestExecuteVerboseParameter:
+    """Test execute() verbose parameter."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_execute_accepts_verbose_parameter(self):
+        """Execute should accept verbose parameter."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(name="step1", action="Action 1")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        def mock_executor(action):
+            return "Result"
+        
+        # Should not raise error
+        result = manager.execute(
+            "test_workflow",
+            executor=mock_executor,
+            verbose=2
+        )
+        
+        assert result["success"]
+    
+    def test_verbose_passed_to_step_agent(self):
+        """Verbose level should be passed to created step agents."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(
+                    name="step1",
+                    action="Action 1",
+                    agent_config={"role": "Helper"}
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            mock_agent = Mock()
+            mock_agent.chat = Mock(return_value="Response")
+            MockAgent.return_value = mock_agent
+            
+            manager.execute(
+                "test_workflow",
+                verbose=3,
+                default_llm="gpt-4o-mini"
+            )
+            
+            call_kwargs = MockAgent.call_args[1]
+            assert "verbose" in call_kwargs
+            assert call_kwargs["verbose"] == 3
+
+
+class TestExecuteResultsWithVariables:
+    """Test that execute() returns variables in results."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_results_include_variables(self):
+        """Results should include accumulated variables."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            variables={"initial": "value"},
+            steps=[
+                WorkflowStep(name="step1", action="Action 1"),
+                WorkflowStep(name="step2", action="Action 2")
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        def mock_executor(action):
+            return f"Output for {action}"
+        
+        result = manager.execute("test_workflow", executor=mock_executor)
+        
+        assert "variables" in result
+        assert "initial" in result["variables"]
+        # Should also have step outputs as variables
+        assert "previous_output" in result["variables"]
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

src/praisonai-agents/tests/test_workflow_tools.py

@@ -0,0 +1,306 @@
+"""
+Test suite for Workflow Per-Step Tools Support.
+
+Tests the enhanced workflow system's ability to configure different tools
+for each step, including tool execution and tool results in context.
+"""
+
+import pytest
+from unittest.mock import Mock, patch
+import sys
+import os
+
+# Add the package to path for testing
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from praisonaiagents.memory.workflows import (
+    WorkflowStep,
+    Workflow,
+    WorkflowManager
+)
+
+
+class TestWorkflowStepToolsField:
+    """Test WorkflowStep tools field."""
+    
+    def test_workflow_step_tools_accepts_list(self):
+        """WorkflowStep should accept a list of tools."""
+        mock_tool1 = Mock()
+        mock_tool2 = Mock()
+        
+        step = WorkflowStep(
+            name="research",
+            action="Research topic",
+            tools=[mock_tool1, mock_tool2]
+        )
+        
+        assert step.tools is not None
+        assert len(step.tools) == 2
+        assert mock_tool1 in step.tools
+        assert mock_tool2 in step.tools
+    
+    def test_workflow_step_tools_accepts_callable(self):
+        """WorkflowStep should accept callable tools."""
+        def search_tool(query):
+            return f"Results for {query}"
+        
+        step = WorkflowStep(
+            name="search",
+            action="Search for info",
+            tools=[search_tool]
+        )
+        
+        assert step.tools is not None
+        assert callable(step.tools[0])
+    
+    def test_workflow_step_tools_in_to_dict(self):
+        """to_dict() should include tools field."""
+        step = WorkflowStep(
+            name="test",
+            action="test action",
+            tools=["tool1", "tool2"]
+        )
+        
+        d = step.to_dict()
+        assert "tools" in d
+        assert d["tools"] == ["tool1", "tool2"]
+
+
+class TestToolsPassedToAgent:
+    """Test that tools are passed to agent during step execution."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_step_tools_passed_to_agent(self):
+        """Step's tools should be passed to created agent."""
+        manager = WorkflowManager()
+        
+        mock_tool = Mock()
+        step = WorkflowStep(
+            name="research",
+            action="Research topic",
+            agent_config={"role": "Researcher"},
+            tools=[mock_tool]
+        )
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            MockAgent.return_value = Mock()
+            manager._create_step_agent(step=step, default_llm="gpt-4o-mini")
+            
+            call_kwargs = MockAgent.call_args[1]
+            assert "tools" in call_kwargs
+            assert mock_tool in call_kwargs["tools"]
+    
+    def test_multiple_tools_passed_to_agent(self):
+        """Multiple tools should all be passed to agent."""
+        manager = WorkflowManager()
+        
+        tool1 = Mock(name="search")
+        tool2 = Mock(name="calculator")
+        tool3 = Mock(name="browser")
+        
+        step = WorkflowStep(
+            name="research",
+            action="Research topic",
+            agent_config={"role": "Researcher"},
+            tools=[tool1, tool2, tool3]
+        )
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            MockAgent.return_value = Mock()
+            manager._create_step_agent(step=step, default_llm="gpt-4o-mini")
+            
+            call_kwargs = MockAgent.call_args[1]
+            assert len(call_kwargs["tools"]) == 3
+
+
+class TestDifferentToolsPerStep:
+    """Test different tools for different steps."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_different_steps_different_tools(self):
+        """Different steps can have different tools."""
+        manager = WorkflowManager()
+        
+        search_tool = Mock(name="search")
+        write_tool = Mock(name="write")
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(
+                    name="research",
+                    action="Research AI",
+                    agent_config={"role": "Researcher"},
+                    tools=[search_tool]
+                ),
+                WorkflowStep(
+                    name="write",
+                    action="Write article",
+                    agent_config={"role": "Writer"},
+                    tools=[write_tool]
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        created_agents = []
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            def track_agent(**kwargs):
+                agent = Mock()
+                agent.chat = Mock(return_value="Response")
+                agent.tools = kwargs.get("tools", [])
+                created_agents.append(kwargs)
+                return agent
+            
+            MockAgent.side_effect = track_agent
+            
+            manager.execute("test_workflow", default_llm="gpt-4o-mini")
+            
+            # Verify different tools for each step
+            assert len(created_agents) == 2
+            assert search_tool in created_agents[0]["tools"]
+            assert write_tool in created_agents[1]["tools"]
+    
+    def test_step_without_tools(self):
+        """Steps without tools should work correctly."""
+        manager = WorkflowManager()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(
+                    name="think",
+                    action="Think about topic",
+                    agent_config={"role": "Thinker"}
+                    # No tools
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            mock_agent = Mock()
+            mock_agent.chat = Mock(return_value="Thoughts")
+            MockAgent.return_value = mock_agent
+            
+            result = manager.execute("test_workflow", default_llm="gpt-4o-mini")
+            
+            assert result["success"]
+            # tools should not be in kwargs or should be None
+            call_kwargs = MockAgent.call_args[1]
+            assert call_kwargs.get("tools") is None or call_kwargs.get("tools") == []
+
+
+class TestToolsWithExecutor:
+    """Test tools behavior when using custom executor."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_executor_ignores_step_tools(self):
+        """When executor is provided, step tools are not used for agent creation."""
+        manager = WorkflowManager()
+        
+        mock_tool = Mock()
+        
+        workflow = Workflow(
+            name="test_workflow",
+            steps=[
+                WorkflowStep(
+                    name="step1",
+                    action="Action 1",
+                    tools=[mock_tool]  # Tools defined but executor provided
+                )
+            ]
+        )
+        self._register_workflow(manager, workflow)
+        
+        executor_called = [False]
+        
+        def custom_executor(action):
+            executor_called[0] = True
+            return "Executor result"
+        
+        # Should not try to create agent since executor is provided
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            result = manager.execute("test_workflow", executor=custom_executor)
+            
+            assert result["success"]
+            assert executor_called[0]
+            MockAgent.assert_not_called()
+
+
+class TestToolsIntegration:
+    """Integration tests for tools in workflow execution."""
+    
+    @staticmethod
+    def _register_workflow(manager, workflow):
+        """Helper to register workflow and mark as loaded."""
+        manager._workflows[workflow.name.lower()] = workflow
+        manager._loaded = True
+    
+    def test_full_workflow_with_tools(self):
+        """Full workflow execution with tools should work."""
+        manager = WorkflowManager()
+        
+        # Create mock tools
+        search_tool = Mock(name="search")
+        analyze_tool = Mock(name="analyze")
+        
+        workflow = Workflow(
+            name="research_workflow",
+            steps=[
+                WorkflowStep(
+                    name="search",
+                    action="Search for {{topic}}",
+                    agent_config={"role": "Searcher"},
+                    tools=[search_tool]
+                ),
+                WorkflowStep(
+                    name="analyze",
+                    action="Analyze: {{previous_output}}",
+                    agent_config={"role": "Analyzer"},
+                    tools=[analyze_tool]
+                ),
+                WorkflowStep(
+                    name="summarize",
+                    action="Summarize findings",
+                    agent_config={"role": "Summarizer"}
+                    # No tools for final step
+                )
+            ],
+            variables={"topic": "AI trends"}
+        )
+        self._register_workflow(manager, workflow)
+        
+        with patch('praisonaiagents.agent.agent.Agent') as MockAgent:
+            def create_mock_agent(**kwargs):
+                agent = Mock()
+                agent.chat = Mock(return_value=f"Response from {kwargs.get('role', 'agent')}")
+                return agent
+            
+            MockAgent.side_effect = create_mock_agent
+            
+            result = manager.execute("research_workflow")
+            
+            assert result["success"]
+            assert len(result["results"]) == 3
+            assert MockAgent.call_count == 3
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

src/praisonai-agents/uv.lock

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

src/praisonai/README.md

@@ -40,7 +40,7 @@ PraisonAI is a production-ready Multi-AI Agents framework with self-reflection,
 | 🔄 Self Reflection AI Agents | [Example](examples/python/concepts/self-reflection-details.py) | [📖](https://docs.praison.ai/features/selfreflection) |
 | 🧠 Reasoning AI Agents | [Example](examples/python/concepts/reasoning-extraction.py) | [📖](https://docs.praison.ai/features/reasoning) |
 | 👁️ Multi Modal AI Agents | [Example](examples/python/general/multimodal.py) | [📖](https://docs.praison.ai/features/multimodal) |
-| 🎭 AI Agent Workflow | [Example](examples/python/stateful/workflow-state-example.py) | [📖](https://docs.praison.ai/features/workflows) |
+| 🎭 AI Agent Workflow (Context Passing, Per-Step Agents) | [Example](examples/python/stateful/workflow-state-example.py) | [📖](https://docs.praison.ai/features/workflows) |
 | 📚 Add Custom Knowledge | [Example](examples/python/concepts/knowledge-agents.py) | [📖](https://docs.praison.ai/features/knowledge) |
 | 🧠 Memory (Short & Long Term) | [Example](examples/python/general/memory_example.py) | [📖](https://docs.praison.ai/concepts/memory) |
 | 📄 Chat with PDF Agents | [Example](examples/python/concepts/chat-with-pdf.py) | [📖](https://docs.praison.ai/features/chat-with-pdf) |
@@ -392,21 +392,53 @@ memories = auto.process_interaction(
 
 ### 8. Workflows
 
-Create reusable multi-step workflows in `.praison/workflows/`:
+Create reusable multi-step workflows with context passing and per-step agents:
 
 ```python
-from praisonaiagents.memory import WorkflowManager
+from praisonaiagents import Agent
+from praisonaiagents.memory import WorkflowManager, Workflow, WorkflowStep
 
+# Simple execution with default agent
+agent = Agent(name="Assistant", llm="gpt-4o-mini")
 manager = WorkflowManager()
 
-# Execute a workflow
 result = manager.execute(
     "deploy",
-    executor=lambda prompt: agent.chat(prompt),
+    default_agent=agent,
     variables={"environment": "production"}
 )
+
+# Advanced: Per-step agent configuration
+workflow = Workflow(
+    name="research_pipeline",
+    default_llm="gpt-4o-mini",
+    steps=[
+        WorkflowStep(
+            name="research",
+            action="Research {{topic}}",
+            agent_config={"role": "Researcher", "goal": "Find information"},
+            tools=["tavily_search"]
+        ),
+        WorkflowStep(
+            name="write",
+            action="Write report based on {{previous_output}}",
+            agent_config={"role": "Writer", "goal": "Write content"},
+            context_from=["research"]  # Only include research output
+        )
+    ]
+)
+
+# Async execution
+import asyncio
+result = asyncio.run(manager.aexecute("deploy", default_llm="gpt-4o-mini"))
 ```
 
+**Key Features:**
+- **Context Passing**: Use `{{previous_output}}` and `{{step_name_output}}` variables
+- **Per-Step Agents**: Configure different agents with roles, goals, tools for each step
+- **Async Execution**: Use `aexecute()` for async workflows
+- **Planning Mode**: Enable at workflow level with `planning=True`
+
 ### 9. Hooks
 
 Configure in `.praison/hooks.json`:

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.13.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.13.tar.gz | shasum -a 256`.split.first
+    url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.14.tar.gz"
+    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.14.tar.gz | shasum -a 256`.split.first
     license "MIT"
   
     depends_on "[email protected]"

src/praisonai/praisonai/cli/main.py

@@ -594,6 +594,7 @@ class PraisonAI:
         parser.add_argument("--planning-reasoning", action="store_true", help="Enable chain-of-thought reasoning in planning")
         parser.add_argument("--auto-approve-plan", action="store_true", help="Auto-approve generated plans without user confirmation")
         parser.add_argument("--max-tokens", type=int, default=16000, help="Maximum output tokens for agent responses (default: 16000)")
+        parser.add_argument("--final-agent", type=str, help="Final agent instruction to process the output (e.g., 'Write a detailed blog post')")
         
         # Memory arguments
         parser.add_argument("--memory", action="store_true", help="Enable file-based memory for agent")
@@ -1933,6 +1934,42 @@ class PraisonAI:
                 flow = FlowDisplayHandler(verbose=getattr(self.args, 'verbose', False))
                 flow.display_workflow_end(success=True)
             
+            # Final Agent - Process output with a specialized agent
+            if hasattr(self, 'args') and getattr(self.args, 'final_agent', None):
+                final_instruction = self.args.final_agent
+                print(f"\n[bold blue]📝 FINAL AGENT PROCESSING[/bold blue]")
+                print(f"[dim]Instruction: {final_instruction}[/dim]\n")
+                
+                # Create a final agent with the same LLM config
+                final_agent_config = {
+                    "name": "FinalAgent",
+                    "role": final_instruction,
+                    "goal": f"Process the provided content and {final_instruction.lower()}",
+                    "backstory": f"You are an expert at {final_instruction.lower()}. You take research content and transform it into polished, detailed output."
+                }
+                
+                # Use same LLM config
+                if agent_config.get("llm"):
+                    final_agent_config["llm"] = agent_config["llm"]
+                
+                final_prompt = f"""Based on the following research content, {final_instruction.lower()}.
+
+## Research Content:
+
+{result}
+
+## Instructions:
+- Be comprehensive and detailed
+- Include all relevant information from the research
+- Structure the output professionally
+- Do not omit any important details
+
+Now, {final_instruction.lower()}:"""
+                
+                final_agent = PraisonAgent(**final_agent_config)
+                result = final_agent.start(final_prompt)
+                print(f"\n[bold green]✅ Final agent processing complete[/bold green]\n")
+            
             # Save output if --save is enabled
             if hasattr(self, 'args') and getattr(self.args, 'save', False):
                 self._save_output(prompt, result)

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.13 gunicorn markdown\n")
+            file.write("RUN pip install flask praisonai==2.3.14 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.13"
+__version__ = "2.3.14"

src/praisonai/pyproject.toml

@@ -12,7 +12,7 @@ dependencies = [
     "rich>=13.7",
     "markdown>=3.5",
     "pyparsing>=3.0.0",
-    "praisonaiagents>=0.0.191",
+    "praisonaiagents>=0.0.192",
     "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.191" },
+    { name = "praisonaiagents", specifier = ">=0.0.192" },
     { 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.191"
+version = "0.0.192"
 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/04/54/431572ac5d5e8efeda19eaace82455d870d3ea9e8d6302d675118512f8a0/praisonaiagents-0.0.191.tar.gz", hash = "sha256:ccb9eee5dd01f926d435c7a9b40c727a9a069697913a51fbe5b00c523334ff9f", size = 492407, upload-time = "2025-12-16T15:01:38.597Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/7f/83/5c3a2bc494a23750571ea5dada6eaafd14fe09eca674bc2483d3659b28d4/praisonaiagents-0.0.192.tar.gz", hash = "sha256:f890f2938a29ff20908e0090537623c9127be9cd1ee8355962e3e5c6f4986e2a", size = 503976, upload-time = "2025-12-16T18:50:30.517Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/0e/06/f20f331e2b34c3970ea01946dd2765ceb8c816cb4b09e8c74c6ad2860e44/praisonaiagents-0.0.191-py3-none-any.whl", hash = "sha256:784e5b7bf6499b9e153591eb619fd5e900bd789ad6adf1e93d2b50cc4ccb2aad", size = 474674, upload-time = "2025-12-16T15:01:37.016Z" },
+    { url = "https://files.pythonhosted.org/packages/a6/22/dc629225b2e5d5e3cf8eb8389490428e9d1ea55575256b99978b4aae255e/praisonaiagents-0.0.192-py3-none-any.whl", hash = "sha256:658378c2235ff5644cb7a294f8ce0a98642e8802a7055ba603e32bdf4f004925", size = 478192, upload-time = "2025-12-16T18:50:28.174Z" },
 ]
 
 [[package]]