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`:
@@ -572,6 +604,362 @@ praisonai hooks init
 praisonai "Research and remember findings" --claude-memory --llm anthropic/claude-sonnet-4-20250514
 ```
 
+### Guardrail CLI:
+```bash
+# Validate output with LLM guardrail
+praisonai "Write code" --guardrail "Ensure code is secure and follows best practices"
+
+# Combine with other flags
+praisonai "Generate SQL query" --guardrail "No DROP or DELETE statements" --save
+```
+
+### Metrics CLI:
+```bash
+# Display token usage and cost metrics
+praisonai "Analyze this data" --metrics
+
+# Combine with other features
+praisonai "Complex task" --metrics --planning
+```
+
+### Image Processing CLI:
+```bash
+# Process images with vision-based tasks
+praisonai "Describe this image" --image path/to/image.png
+
+# Analyze image content
+praisonai "What objects are in this photo?" --image photo.jpg --llm openai/gpt-4o
+```
+
+### Telemetry CLI:
+```bash
+# Enable usage monitoring and analytics
+praisonai "Task" --telemetry
+
+# Combine with metrics for full observability
+praisonai "Complex analysis" --telemetry --metrics
+```
+
+### MCP (Model Context Protocol) CLI:
+```bash
+# Use MCP server tools
+praisonai "Search files" --mcp "npx -y @modelcontextprotocol/server-filesystem ."
+
+# MCP with environment variables
+praisonai "Search web" --mcp "npx -y @modelcontextprotocol/server-brave-search" --mcp-env "BRAVE_API_KEY=your_key"
+
+# Multiple MCP options
+praisonai "Task" --mcp "npx server" --mcp-env "KEY1=value1,KEY2=value2"
+```
+
+### Fast Context CLI:
+```bash
+# Search codebase for relevant context
+praisonai "Find authentication code" --fast-context ./src
+
+# Add code context to any task
+praisonai "Explain this function" --fast-context /path/to/project
+```
+
+### Knowledge CLI:
+```bash
+# Add documents to knowledge base
+praisonai knowledge add document.pdf
+praisonai knowledge add ./docs/
+
+# Search knowledge base
+praisonai knowledge search "API authentication"
+
+# List indexed documents
+praisonai knowledge list
+
+# Clear knowledge base
+praisonai knowledge clear
+
+# Show knowledge base info
+praisonai knowledge info
+
+# Show all commands
+praisonai knowledge help
+```
+
+### Session CLI:
+```bash
+# Start a new session
+praisonai session start my-project
+
+# List all sessions
+praisonai session list
+
+# Resume a session
+praisonai session resume my-project
+
+# Show session details
+praisonai session show my-project
+
+# Delete a session
+praisonai session delete my-project
+
+# Show all commands
+praisonai session help
+```
+
+### Tools CLI:
+```bash
+# List all available tools
+praisonai tools list
+
+# Get info about a specific tool
+praisonai tools info internet_search
+
+# Search for tools
+praisonai tools search "web"
+
+# Show all commands
+praisonai tools help
+```
+
+### Handoff CLI:
+```bash
+# Enable agent-to-agent task delegation
+praisonai "Research and write article" --handoff "researcher,writer,editor"
+
+# Complex multi-agent workflow
+praisonai "Analyze data and create report" --handoff "analyst,visualizer,writer"
+```
+
+### Auto Memory CLI:
+```bash
+# Enable automatic memory extraction
+praisonai "Learn about user preferences" --auto-memory
+
+# Combine with user isolation
+praisonai "Remember my settings" --auto-memory --user-id user123
+```
+
+### Todo CLI:
+```bash
+# Generate todo list from task
+praisonai "Plan the project" --todo
+
+# Add a todo item
+praisonai todo add "Implement feature X"
+
+# List all todos
+praisonai todo list
+
+# Complete a todo
+praisonai todo complete 1
+
+# Delete a todo
+praisonai todo delete 1
+
+# Clear all todos
+praisonai todo clear
+
+# Show all commands
+praisonai todo help
+```
+
+### Router CLI:
+```bash
+# Auto-select best model based on task complexity
+praisonai "Simple question" --router
+
+# Specify preferred provider
+praisonai "Complex analysis" --router --router-provider anthropic
+
+# Router automatically selects:
+# - Simple tasks → gpt-4o-mini, claude-3-haiku
+# - Complex tasks → gpt-4-turbo, claude-3-opus
+```
+
+### Flow Display CLI:
+```bash
+# Enable visual workflow tracking
+praisonai agents.yaml --flow-display
+
+# Combine with other features
+praisonai "Multi-step task" --planning --flow-display
+```
+
+## Prompt Expansion
+
+Expand short prompts into detailed, actionable prompts:
+
+### CLI Usage
+```bash
+# Expand a short prompt into detailed prompt
+praisonai "write a movie script in 3 lines" --expand-prompt
+
+# With verbose output
+praisonai "blog about AI" --expand-prompt -v
+
+# With tools for context gathering
+praisonai "latest AI trends" --expand-prompt --expand-tools tools.py
+
+# Combine with query rewrite
+praisonai "AI news" --query-rewrite --expand-prompt
+```
+
+### Programmatic Usage
+```python
+from praisonaiagents import PromptExpanderAgent, ExpandStrategy
+
+# Basic usage
+agent = PromptExpanderAgent()
+result = agent.expand("write a movie script in 3 lines")
+print(result.expanded_prompt)
+
+# With specific strategy
+result = agent.expand("blog about AI", strategy=ExpandStrategy.DETAILED)
+
+# Available strategies: BASIC, DETAILED, STRUCTURED, CREATIVE, AUTO
+```
+
+**Key Difference:**
+- `--query-rewrite`: Optimizes queries for search/retrieval (RAG)
+- `--expand-prompt`: Expands prompts for detailed task execution
+
+## Web Search, Web Fetch & Prompt Caching
+
+### CLI Usage
+```bash
+# Web Search - Get real-time information
+praisonai "What are the latest AI news today?" --web-search --llm openai/gpt-4o-search-preview
+
+# Web Fetch - Retrieve and analyze URL content (Anthropic only)
+praisonai "Summarize https://docs.praison.ai" --web-fetch --llm anthropic/claude-sonnet-4-20250514
+
+# Prompt Caching - Reduce costs for repeated prompts
+praisonai "Analyze this document..." --prompt-caching --llm anthropic/claude-sonnet-4-20250514
+```
+
+### Programmatic Usage
+```python
+from praisonaiagents import Agent
+
+# Web Search
+agent = Agent(
+    instructions="You are a research assistant",
+    llm="openai/gpt-4o-search-preview",
+    web_search=True
+)
+
+# Web Fetch (Anthropic only)
+agent = Agent(
+    instructions="You are a content analyzer",
+    llm="anthropic/claude-sonnet-4-20250514",
+    web_fetch=True
+)
+
+# Prompt Caching
+agent = Agent(
+    instructions="You are an AI assistant..." * 50,  # Long system prompt
+    llm="anthropic/claude-sonnet-4-20250514",
+    prompt_caching=True
+)
+```
+
+**Supported Providers:**
+| Feature | Providers |
+|---------|----------|
+| Web Search | OpenAI, Gemini, Anthropic, xAI, Perplexity |
+| Web Fetch | Anthropic |
+| Prompt Caching | OpenAI (auto), Anthropic, Bedrock, Deepseek |
+
+## MCP (Model Context Protocol)
+
+PraisonAI supports MCP Protocol Revision 2025-11-25 with multiple transports.
+
+### MCP Client (Consume MCP Servers)
+```python
+from praisonaiagents import Agent, MCP
+
+# stdio - Local NPX/Python servers
+agent = Agent(tools=MCP("npx @modelcontextprotocol/server-memory"))
+
+# Streamable HTTP - Production servers
+agent = Agent(tools=MCP("https://api.example.com/mcp"))
+
+# WebSocket - Real-time bidirectional
+agent = Agent(tools=MCP("wss://api.example.com/mcp", auth_token="token"))
+
+# SSE (Legacy) - Backward compatibility
+agent = Agent(tools=MCP("http://localhost:8080/sse"))
+
+# With environment variables
+agent = Agent(
+    tools=MCP(
+        command="npx",
+        args=["-y", "@modelcontextprotocol/server-brave-search"],
+        env={"BRAVE_API_KEY": "your-key"}
+    )
+)
+```
+
+### MCP Server (Expose Tools as MCP Server)
+
+Expose your Python functions as MCP tools for Claude Desktop, Cursor, and other MCP clients:
+
+```python
+from praisonaiagents.mcp import ToolsMCPServer
+
+def search_web(query: str, max_results: int = 5) -> dict:
+    """Search the web for information."""
+    return {"results": [f"Result for {query}"]}
+
+def calculate(expression: str) -> dict:
+    """Evaluate a mathematical expression."""
+    return {"result": eval(expression)}
+
+# Create and run MCP server
+server = ToolsMCPServer(name="my-tools")
+server.register_tools([search_web, calculate])
+server.run()  # stdio for Claude Desktop
+# server.run_sse(host="0.0.0.0", port=8080)  # SSE for web clients
+```
+
+### MCP Features
+| Feature | Description |
+|---------|-------------|
+| Session Management | Automatic Mcp-Session-Id handling |
+| Protocol Versioning | Mcp-Protocol-Version header |
+| Resumability | SSE stream recovery via Last-Event-ID |
+| Security | Origin validation, DNS rebinding prevention |
+| WebSocket | Auto-reconnect with exponential backoff |
+
+## CLI Features
+
+| Feature | Docs |
+|---------|:----:|
+| 🔄 Query Rewrite - RAG optimization | [📖](https://docs.praison.ai/cli/query-rewrite) |
+| 🔬 Deep Research - Automated research | [📖](https://docs.praison.ai/cli/deep-research) |
+| 📋 Planning - Step-by-step execution | [📖](https://docs.praison.ai/cli/planning) |
+| 💾 Memory - Persistent agent memory | [📖](https://docs.praison.ai/cli/memory) |
+| 📜 Rules - Auto-discovered instructions | [📖](https://docs.praison.ai/cli/rules) |
+| 🔄 Workflow - Multi-step workflows | [📖](https://docs.praison.ai/cli/workflow) |
+| 🪝 Hooks - Event-driven actions | [📖](https://docs.praison.ai/cli/hooks) |
+| 🧠 Claude Memory - Anthropic memory tool | [📖](https://docs.praison.ai/cli/claude-memory) |
+| 🛡️ Guardrail - Output validation | [📖](https://docs.praison.ai/cli/guardrail) |
+| 📊 Metrics - Token usage tracking | [📖](https://docs.praison.ai/cli/metrics) |
+| 🖼️ Image - Vision processing | [📖](https://docs.praison.ai/cli/image) |
+| 📡 Telemetry - Usage monitoring | [📖](https://docs.praison.ai/cli/telemetry) |
+| 🔌 MCP - Model Context Protocol | [📖](https://docs.praison.ai/cli/mcp) |
+| ⚡ Fast Context - Codebase search | [📖](https://docs.praison.ai/cli/fast-context) |
+| 📚 Knowledge - RAG management | [📖](https://docs.praison.ai/cli/knowledge) |
+| 💬 Session - Conversation management | [📖](https://docs.praison.ai/cli/session) |
+| 🔧 Tools - Tool discovery | [📖](https://docs.praison.ai/cli/tools) |
+| 🤝 Handoff - Agent delegation | [📖](https://docs.praison.ai/cli/handoff) |
+| 🧠 Auto Memory - Memory extraction | [📖](https://docs.praison.ai/cli/auto-memory) |
+| 📋 Todo - Task management | [📖](https://docs.praison.ai/cli/todo) |
+| 🎯 Router - Smart model selection | [📖](https://docs.praison.ai/cli/router) |
+| 📈 Flow Display - Visual workflow | [📖](https://docs.praison.ai/cli/flow-display) |
+| ✨ Prompt Expansion - Detailed prompts | [📖](https://docs.praison.ai/cli/prompt-expansion) |
+| 🌐 Web Search - Real-time search | [📖](https://docs.praison.ai/cli/web-search) |
+| 📥 Web Fetch - URL content retrieval | [📖](https://docs.praison.ai/cli/web-fetch) |
+| 💾 Prompt Caching - Cost reduction | [📖](https://docs.praison.ai/cli/prompt-caching) |
+
 ## Using JavaScript Code
 
 ```bash

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

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

@@ -2627,10 +2627,10 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
             
             # Execute the step
             result = self.chat(step_prompt, **kwargs)
-            results.append(result)
+            results.append({"step": i + 1, "description": step.description, "result": result})
             
-            # Update context for next step
-            context += f"\n\nStep {i + 1} result: {result[:500] if result else 'No result'}"
+            # Update context for next step (use full result, not truncated)
+            context += f"\n\nStep {i + 1} result: {result if result else 'No result'}"
             
             console.print(f"   [green]✅ Completed[/green]")
         
@@ -2638,8 +2638,41 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
         console.print(f"[dim]Progress: [{'█' * bar_length}] 100%[/dim]")
         console.print(f"Completed {len(plan.steps)}/{len(plan.steps)} steps!\n")
         
-        # Return the final result
-        return results[-1] if results else None
+        # Compile all results into a comprehensive final output
+        if len(results) > 1:
+            # Create a compilation prompt
+            all_results_text = "\n\n".join([
+                f"## Step {r['step']}: {r['description']}\n\n{r['result']}" 
+                for r in results
+            ])
+            
+            compilation_prompt = f"""You are tasked with compiling a comprehensive, detailed report from the following research steps.
+
+IMPORTANT: Write a DETAILED and COMPREHENSIVE report. Do NOT summarize or compress the information. 
+Include ALL relevant details, data, statistics, and findings from each step.
+Organize the information logically with clear sections and subsections.
+
+## Research Results to Compile:
+
+{all_results_text}
+
+## Instructions:
+1. Combine all the information into a single, well-organized document
+2. Preserve ALL details, numbers, statistics, and specific findings
+3. Use clear headings and subheadings
+4. Do not omit any important information
+5. Make it comprehensive and detailed
+
+Write the complete compiled report:"""
+            
+            console.print("\n[bold blue]📝 COMPILING FINAL REPORT[/bold blue]")
+            console.print("[dim]Creating comprehensive output from all steps...[/dim]\n")
+            
+            final_result = self.chat(compilation_prompt, **kwargs)
+            return final_result
+        
+        # Return the single result if only one step
+        return results[0]["result"] if results else None
 
     def start(self, prompt: str, **kwargs):
         """Start the agent with a prompt. This is a convenience method that wraps chat()."""

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

@@ -347,6 +347,37 @@ Expanded prompt:"""
             current_date=current_date
         )
         
+        # Add tool usage instructions if tools are available
+        if self.tools:
+            # Build dynamic tool descriptions from docstrings
+            tool_descriptions = []
+            for tool in self.tools:
+                tool_name = getattr(tool, '__name__', str(tool))
+                tool_doc = getattr(tool, '__doc__', '')
+                # Get first line of docstring as description
+                if tool_doc:
+                    first_line = tool_doc.strip().split('\n')[0].strip()
+                    tool_descriptions.append(f"- {tool_name}: {first_line}")
+                else:
+                    tool_descriptions.append(f"- {tool_name}")
+            
+            tools_list = '\n'.join(tool_descriptions)
+            
+            tool_instruction = f"""
+
+IMPORTANT: You have the following tools available:
+{tools_list}
+
+Before expanding this prompt, FIRST use the available tools to gather relevant current information about the topic. This will help you create a more informed and accurate expanded prompt.
+
+Steps:
+1. Use the appropriate tool(s) to gather relevant information about the topic
+2. Use the gathered information to inform your expansion
+3. Create an expanded prompt that incorporates the latest context
+
+"""
+            expansion_prompt = tool_instruction + expansion_prompt
+        
         # Call the agent
         expanded = self._call_agent(expansion_prompt).strip()
         

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`:
@@ -572,6 +604,362 @@ praisonai hooks init
 praisonai "Research and remember findings" --claude-memory --llm anthropic/claude-sonnet-4-20250514
 ```
 
+### Guardrail CLI:
+```bash
+# Validate output with LLM guardrail
+praisonai "Write code" --guardrail "Ensure code is secure and follows best practices"
+
+# Combine with other flags
+praisonai "Generate SQL query" --guardrail "No DROP or DELETE statements" --save
+```
+
+### Metrics CLI:
+```bash
+# Display token usage and cost metrics
+praisonai "Analyze this data" --metrics
+
+# Combine with other features
+praisonai "Complex task" --metrics --planning
+```
+
+### Image Processing CLI:
+```bash
+# Process images with vision-based tasks
+praisonai "Describe this image" --image path/to/image.png
+
+# Analyze image content
+praisonai "What objects are in this photo?" --image photo.jpg --llm openai/gpt-4o
+```
+
+### Telemetry CLI:
+```bash
+# Enable usage monitoring and analytics
+praisonai "Task" --telemetry
+
+# Combine with metrics for full observability
+praisonai "Complex analysis" --telemetry --metrics
+```
+
+### MCP (Model Context Protocol) CLI:
+```bash
+# Use MCP server tools
+praisonai "Search files" --mcp "npx -y @modelcontextprotocol/server-filesystem ."
+
+# MCP with environment variables
+praisonai "Search web" --mcp "npx -y @modelcontextprotocol/server-brave-search" --mcp-env "BRAVE_API_KEY=your_key"
+
+# Multiple MCP options
+praisonai "Task" --mcp "npx server" --mcp-env "KEY1=value1,KEY2=value2"
+```
+
+### Fast Context CLI:
+```bash
+# Search codebase for relevant context
+praisonai "Find authentication code" --fast-context ./src
+
+# Add code context to any task
+praisonai "Explain this function" --fast-context /path/to/project
+```
+
+### Knowledge CLI:
+```bash
+# Add documents to knowledge base
+praisonai knowledge add document.pdf
+praisonai knowledge add ./docs/
+
+# Search knowledge base
+praisonai knowledge search "API authentication"
+
+# List indexed documents
+praisonai knowledge list
+
+# Clear knowledge base
+praisonai knowledge clear
+
+# Show knowledge base info
+praisonai knowledge info
+
+# Show all commands
+praisonai knowledge help
+```
+
+### Session CLI:
+```bash
+# Start a new session
+praisonai session start my-project
+
+# List all sessions
+praisonai session list
+
+# Resume a session
+praisonai session resume my-project
+
+# Show session details
+praisonai session show my-project
+
+# Delete a session
+praisonai session delete my-project
+
+# Show all commands
+praisonai session help
+```
+
+### Tools CLI:
+```bash
+# List all available tools
+praisonai tools list
+
+# Get info about a specific tool
+praisonai tools info internet_search
+
+# Search for tools
+praisonai tools search "web"
+
+# Show all commands
+praisonai tools help
+```
+
+### Handoff CLI:
+```bash
+# Enable agent-to-agent task delegation
+praisonai "Research and write article" --handoff "researcher,writer,editor"
+
+# Complex multi-agent workflow
+praisonai "Analyze data and create report" --handoff "analyst,visualizer,writer"
+```
+
+### Auto Memory CLI:
+```bash
+# Enable automatic memory extraction
+praisonai "Learn about user preferences" --auto-memory
+
+# Combine with user isolation
+praisonai "Remember my settings" --auto-memory --user-id user123
+```
+
+### Todo CLI:
+```bash
+# Generate todo list from task
+praisonai "Plan the project" --todo
+
+# Add a todo item
+praisonai todo add "Implement feature X"
+
+# List all todos
+praisonai todo list
+
+# Complete a todo
+praisonai todo complete 1
+
+# Delete a todo
+praisonai todo delete 1
+
+# Clear all todos
+praisonai todo clear
+
+# Show all commands
+praisonai todo help
+```
+
+### Router CLI:
+```bash
+# Auto-select best model based on task complexity
+praisonai "Simple question" --router
+
+# Specify preferred provider
+praisonai "Complex analysis" --router --router-provider anthropic
+
+# Router automatically selects:
+# - Simple tasks → gpt-4o-mini, claude-3-haiku
+# - Complex tasks → gpt-4-turbo, claude-3-opus
+```
+
+### Flow Display CLI:
+```bash
+# Enable visual workflow tracking
+praisonai agents.yaml --flow-display
+
+# Combine with other features
+praisonai "Multi-step task" --planning --flow-display
+```
+
+## Prompt Expansion
+
+Expand short prompts into detailed, actionable prompts:
+
+### CLI Usage
+```bash
+# Expand a short prompt into detailed prompt
+praisonai "write a movie script in 3 lines" --expand-prompt
+
+# With verbose output
+praisonai "blog about AI" --expand-prompt -v
+
+# With tools for context gathering
+praisonai "latest AI trends" --expand-prompt --expand-tools tools.py
+
+# Combine with query rewrite
+praisonai "AI news" --query-rewrite --expand-prompt
+```
+
+### Programmatic Usage
+```python
+from praisonaiagents import PromptExpanderAgent, ExpandStrategy
+
+# Basic usage
+agent = PromptExpanderAgent()
+result = agent.expand("write a movie script in 3 lines")
+print(result.expanded_prompt)
+
+# With specific strategy
+result = agent.expand("blog about AI", strategy=ExpandStrategy.DETAILED)
+
+# Available strategies: BASIC, DETAILED, STRUCTURED, CREATIVE, AUTO
+```
+
+**Key Difference:**
+- `--query-rewrite`: Optimizes queries for search/retrieval (RAG)
+- `--expand-prompt`: Expands prompts for detailed task execution
+
+## Web Search, Web Fetch & Prompt Caching
+
+### CLI Usage
+```bash
+# Web Search - Get real-time information
+praisonai "What are the latest AI news today?" --web-search --llm openai/gpt-4o-search-preview
+
+# Web Fetch - Retrieve and analyze URL content (Anthropic only)
+praisonai "Summarize https://docs.praison.ai" --web-fetch --llm anthropic/claude-sonnet-4-20250514
+
+# Prompt Caching - Reduce costs for repeated prompts
+praisonai "Analyze this document..." --prompt-caching --llm anthropic/claude-sonnet-4-20250514
+```
+
+### Programmatic Usage
+```python
+from praisonaiagents import Agent
+
+# Web Search
+agent = Agent(
+    instructions="You are a research assistant",
+    llm="openai/gpt-4o-search-preview",
+    web_search=True
+)
+
+# Web Fetch (Anthropic only)
+agent = Agent(
+    instructions="You are a content analyzer",
+    llm="anthropic/claude-sonnet-4-20250514",
+    web_fetch=True
+)
+
+# Prompt Caching
+agent = Agent(
+    instructions="You are an AI assistant..." * 50,  # Long system prompt
+    llm="anthropic/claude-sonnet-4-20250514",
+    prompt_caching=True
+)
+```
+
+**Supported Providers:**
+| Feature | Providers |
+|---------|----------|
+| Web Search | OpenAI, Gemini, Anthropic, xAI, Perplexity |
+| Web Fetch | Anthropic |
+| Prompt Caching | OpenAI (auto), Anthropic, Bedrock, Deepseek |
+
+## MCP (Model Context Protocol)
+
+PraisonAI supports MCP Protocol Revision 2025-11-25 with multiple transports.
+
+### MCP Client (Consume MCP Servers)
+```python
+from praisonaiagents import Agent, MCP
+
+# stdio - Local NPX/Python servers
+agent = Agent(tools=MCP("npx @modelcontextprotocol/server-memory"))
+
+# Streamable HTTP - Production servers
+agent = Agent(tools=MCP("https://api.example.com/mcp"))
+
+# WebSocket - Real-time bidirectional
+agent = Agent(tools=MCP("wss://api.example.com/mcp", auth_token="token"))
+
+# SSE (Legacy) - Backward compatibility
+agent = Agent(tools=MCP("http://localhost:8080/sse"))
+
+# With environment variables
+agent = Agent(
+    tools=MCP(
+        command="npx",
+        args=["-y", "@modelcontextprotocol/server-brave-search"],
+        env={"BRAVE_API_KEY": "your-key"}
+    )
+)
+```
+
+### MCP Server (Expose Tools as MCP Server)
+
+Expose your Python functions as MCP tools for Claude Desktop, Cursor, and other MCP clients:
+
+```python
+from praisonaiagents.mcp import ToolsMCPServer
+
+def search_web(query: str, max_results: int = 5) -> dict:
+    """Search the web for information."""
+    return {"results": [f"Result for {query}"]}
+
+def calculate(expression: str) -> dict:
+    """Evaluate a mathematical expression."""
+    return {"result": eval(expression)}
+
+# Create and run MCP server
+server = ToolsMCPServer(name="my-tools")
+server.register_tools([search_web, calculate])
+server.run()  # stdio for Claude Desktop
+# server.run_sse(host="0.0.0.0", port=8080)  # SSE for web clients
+```
+
+### MCP Features
+| Feature | Description |
+|---------|-------------|
+| Session Management | Automatic Mcp-Session-Id handling |
+| Protocol Versioning | Mcp-Protocol-Version header |
+| Resumability | SSE stream recovery via Last-Event-ID |
+| Security | Origin validation, DNS rebinding prevention |
+| WebSocket | Auto-reconnect with exponential backoff |
+
+## CLI Features
+
+| Feature | Docs |
+|---------|:----:|
+| 🔄 Query Rewrite - RAG optimization | [📖](https://docs.praison.ai/cli/query-rewrite) |
+| 🔬 Deep Research - Automated research | [📖](https://docs.praison.ai/cli/deep-research) |
+| 📋 Planning - Step-by-step execution | [📖](https://docs.praison.ai/cli/planning) |
+| 💾 Memory - Persistent agent memory | [📖](https://docs.praison.ai/cli/memory) |
+| 📜 Rules - Auto-discovered instructions | [📖](https://docs.praison.ai/cli/rules) |
+| 🔄 Workflow - Multi-step workflows | [📖](https://docs.praison.ai/cli/workflow) |
+| 🪝 Hooks - Event-driven actions | [📖](https://docs.praison.ai/cli/hooks) |
+| 🧠 Claude Memory - Anthropic memory tool | [📖](https://docs.praison.ai/cli/claude-memory) |
+| 🛡️ Guardrail - Output validation | [📖](https://docs.praison.ai/cli/guardrail) |
+| 📊 Metrics - Token usage tracking | [📖](https://docs.praison.ai/cli/metrics) |
+| 🖼️ Image - Vision processing | [📖](https://docs.praison.ai/cli/image) |
+| 📡 Telemetry - Usage monitoring | [📖](https://docs.praison.ai/cli/telemetry) |
+| 🔌 MCP - Model Context Protocol | [📖](https://docs.praison.ai/cli/mcp) |
+| ⚡ Fast Context - Codebase search | [📖](https://docs.praison.ai/cli/fast-context) |
+| 📚 Knowledge - RAG management | [📖](https://docs.praison.ai/cli/knowledge) |
+| 💬 Session - Conversation management | [📖](https://docs.praison.ai/cli/session) |
+| 🔧 Tools - Tool discovery | [📖](https://docs.praison.ai/cli/tools) |
+| 🤝 Handoff - Agent delegation | [📖](https://docs.praison.ai/cli/handoff) |
+| 🧠 Auto Memory - Memory extraction | [📖](https://docs.praison.ai/cli/auto-memory) |
+| 📋 Todo - Task management | [📖](https://docs.praison.ai/cli/todo) |
+| 🎯 Router - Smart model selection | [📖](https://docs.praison.ai/cli/router) |
+| 📈 Flow Display - Visual workflow | [📖](https://docs.praison.ai/cli/flow-display) |
+| ✨ Prompt Expansion - Detailed prompts | [📖](https://docs.praison.ai/cli/prompt-expansion) |
+| 🌐 Web Search - Real-time search | [📖](https://docs.praison.ai/cli/web-search) |
+| 📥 Web Fetch - URL content retrieval | [📖](https://docs.praison.ai/cli/web-fetch) |
+| 💾 Prompt Caching - Cost reduction | [📖](https://docs.praison.ai/cli/prompt-caching) |
+
 ## Using JavaScript Code
 
 ```bash

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.12.tar.gz"
-    sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.12.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

@@ -593,6 +593,8 @@ class PraisonAI:
         parser.add_argument("--planning-tools", type=str, help="Tools for planning research (path to tools.py or comma-separated tool names)")
         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")
@@ -1727,7 +1729,14 @@ class PraisonAI:
             
             # Add llm if specified
             if hasattr(self, 'args') and self.args.llm:
-                agent_config["llm"] = self.args.llm
+                # Check if max_tokens is specified - pass as dict config
+                max_tokens = getattr(self.args, 'max_tokens', 16000)
+                if max_tokens:
+                    # Pass llm as dict with model and max_tokens
+                    agent_config["llm"] = {"model": self.args.llm, "max_tokens": max_tokens}
+                    print(f"[bold cyan]Max tokens set to: {max_tokens}[/bold cyan]")
+                else:
+                    agent_config["llm"] = self.args.llm
             
             # Add feature flags if enabled
             if hasattr(self, 'args'):
@@ -1738,6 +1747,18 @@ class PraisonAI:
                 if getattr(self.args, 'prompt_caching', False):
                     agent_config["prompt_caching"] = True
                 
+                # Load tools if specified (--tools flag)
+                if getattr(self.args, 'tools', None):
+                    tools_list = self._load_tools(self.args.tools)
+                    if tools_list:
+                        existing_tools = agent_config.get('tools', [])
+                        if isinstance(existing_tools, list):
+                            existing_tools.extend(tools_list)
+                        else:
+                            existing_tools = tools_list
+                        agent_config['tools'] = existing_tools
+                        print(f"[bold cyan]Tools loaded: {len(tools_list)} tool(s) available for agent[/bold cyan]")
+                
                 # Planning Mode
                 if getattr(self.args, 'planning', False):
                     agent_config["planning"] = True
@@ -1748,6 +1769,10 @@ class PraisonAI:
                         planning_tools_list = self._load_tools(self.args.planning_tools)
                         if planning_tools_list:
                             agent_config["planning_tools"] = planning_tools_list
+                    # If no planning_tools but --tools is specified, use those for planning too
+                    elif getattr(self.args, 'tools', None) and agent_config.get('tools'):
+                        agent_config["planning_tools"] = agent_config['tools']
+                        print("[cyan]Using --tools for planning as well[/cyan]")
                     
                     if getattr(self.args, 'planning_reasoning', False):
                         agent_config["planning_reasoning"] = True
@@ -1909,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.12 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.12"
+__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]]