Release v4.6.55

* fix: prevent duplicate session messages from persist+auto_save and save_state

When memory=history and auto_save share a session_id, per-turn _persist_message
wrote messages that _auto_save_session then re-appended because
_auto_save_last_index was not updated.

Session.save_state() also re-appended the full chat history on every call
instead of replacing it atomically.

- Sync _auto_save_last_index in _persist_message when sessions match
- Use set_chat_history() in Session._save_agent_chat_histories for idempotent saves
- Add regression tests for both duplicate paths

Co-authored-by: Mervin Praison <[email protected]>

* fix: address thread safety and fallback logging issues

- Wrap _auto_save_last_index increment in _history_lock for thread safety
- Add warning when session store lacks set_chat_history (duplicate risk)
- Add debug logging for empty message lists

Co-authored-by: Mervin Praison <[email protected]>

---------

Co-authored-by: Cursor Agent <[email protected]>
Co-authored-by: Mervin Praison <[email protected]>
Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>

.env.example

@@ -0,0 +1,51 @@
+# PraisonAI Environment Variables
+# Copy this file to .env and update with your actual values
+
+# === Required ===
+OPENAI_API_KEY=sk-...          # OpenAI key for LLM calls. Get yours at https://platform.openai.com
+
+# === Required for Claw web-search tool ===
+TAVILY_API_KEY=tvly-...        # Get yours at https://app.tavily.com (free tier available)
+
+# === Optional: Claw dashboard security ===
+PRAISONAI_ALLOW_LOCAL_TOOLS=0  # Set to 1 to enable local shell tools (security risk)
+
+# === Optional: Alternative LLM Providers ===
+ANTHROPIC_API_KEY=sk-ant-...   # Anthropic Claude models
+GOOGLE_API_KEY=...             # Google Gemini models  
+GROQ_API_KEY=gsk_...           # Groq (fast inference)
+COHERE_API_KEY=...             # Cohere models
+AZURE_OPENAI_API_KEY=...       # Azure OpenAI
+AZURE_OPENAI_ENDPOINT=...      # Azure OpenAI endpoint
+
+# === Optional: Memory backends ===
+MEM0_API_KEY=...               # Mem0 memory service
+REDIS_URL=redis://localhost:6379  # Redis for state management
+
+# === Optional: Observability ===
+LANGFUSE_SECRET_KEY=sk-lf-...  # Langfuse tracing
+LANGFUSE_PUBLIC_KEY=pk-lf-...  # Langfuse public key
+
+# === Optional: Bot platforms ===
+TELEGRAM_BOT_TOKEN=...         # Telegram bot token
+DISCORD_BOT_TOKEN=...          # Discord bot token  
+SLACK_BOT_TOKEN=xoxb-...       # Slack bot token
+SLACK_APP_TOKEN=xapp-...       # Slack app token
+WHATSAPP_ACCESS_TOKEN=...      # WhatsApp bot access token
+WHATSAPP_PHONE_NUMBER_ID=...   # WhatsApp phone number ID
+
+# === Optional: Database ===
+DATABASE_URL=sqlite:///~/.praison/database.sqlite  # Database connection string
+
+# === Optional: Tool Security ===
+# ALLOWED_TOOLS=search,send_message       # Whitelist specific tools to prevent name collisions
+                                        # Useful in multi-environment setups with overlapping tool names
+                                        # Leave unset to use all tools (with collision warnings)
+                                        # Do NOT set to empty string: that's treated as configuration error
+                                        # In CI, unknown tool names in the whitelist cause strict startup failure
+# HERMES_ONLY_TOOLS=search,send_message  # (Deprecated) Backward compatibility alias for ALLOWED_TOOLS
+
+# === Optional: Development ===
+LOGLEVEL=INFO                  # Logging level (DEBUG, INFO, WARNING, ERROR)
+CHAINLIT_AUTH_SECRET=...       # Chainlit authentication secret
+DEBUG=false                    # Enable debug mode

.github/workflows/claude.yml

@@ -181,12 +181,15 @@ jobs:
 
       - name: Fetch PR branch and Setup Remote
         if: github.event.issue.pull_request
+        env:
+          PR_BRANCH: ${{ steps.check_fork.outputs.pr_branch }}
+          IS_FORK: ${{ steps.check_fork.outputs.is_fork }}
         run: |
           # Fetch PR head from base repo and put it into local branch
-          git fetch origin pull/${{ github.event.issue.number }}/head:${{ steps.check_fork.outputs.pr_branch }}
+          git fetch origin pull/${{ github.event.issue.number }}/head:"${PR_BRANCH}"
           
           # If it's a fork, make origin point to local to trick claude-code-action's `git fetch origin <branch>`
-          if [ "${{ steps.check_fork.outputs.is_fork }}" == "true" ]; then
+          if [ "${IS_FORK}" = "true" ]; then
             git remote set-url origin file://$(pwd)
           fi
 

.github/workflows/integration-cross-repo.yml

@@ -0,0 +1,158 @@
+name: Cross-Repo Integration Tests
+
+on:
+  push:
+    branches: [ main ]
+    paths:
+      - 'src/praisonai/praisonai/integration/**'
+      - 'src/praisonai/tests/integration/test_aiui_*'
+      - 'src/praisonai/pyproject.toml'
+  pull_request:
+    branches: [ main ]
+    paths:
+      - 'src/praisonai/praisonai/integration/**'
+      - 'src/praisonai/tests/integration/test_aiui_*'
+      - 'src/praisonai/pyproject.toml'
+  workflow_dispatch:
+
+jobs:
+  cross-repo-integration:
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    
+    steps:
+    - name: Checkout PraisonAI
+      uses: actions/checkout@v4
+      with:
+        path: PraisonAI
+        ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
+        
+    - name: Checkout PraisonAIUI
+      uses: actions/checkout@v4
+      with:
+        repository: MervinPraison/PraisonAIUI
+        path: PraisonAIUI
+        ref: main
+        
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: '3.11'
+        
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        # Install PraisonAI packages in development mode
+        cd PraisonAI/src/praisonai-agents
+        pip install -e .
+        cd ../praisonai
+        pip install -e ".[ui,dev]"
+        # Install PraisonAIUI in development mode
+        cd ../../../PraisonAIUI
+        pip install -e .
+        
+    - name: Verify versions
+      run: |
+        python -c "import praisonaiagents; print(f'praisonaiagents: {praisonaiagents.__version__}')"
+        python -c "import praisonai; print(f'praisonai: {praisonai.__version__}')"
+        python -c "import praisonaiui; print(f'praisonaiui: {praisonaiui.__version__}')"
+        
+    - name: Run PraisonAI integration tests
+      run: |
+        cd PraisonAI/src/praisonai
+        python -m pytest tests/integration/test_aiui_* -v --timeout=300
+        
+    - name: Run PraisonAIUI integration tests
+      run: |
+        cd PraisonAIUI
+        if [ -f tests/integration/test_agentic_roundtrip.py ]; then
+          python -m pytest tests/integration/test_agentic_roundtrip.py -v --timeout=300
+        fi
+        if [ -f tests/test_feature_sdk_backends.py ]; then
+          python -m pytest tests/test_feature_sdk_backends.py -v --timeout=300
+        fi
+        
+    - name: Test Pattern C CLI
+      run: |
+        cd PraisonAI/src/praisonai
+        # Test that ui-gateway command exists
+        python -m praisonai serve --help | grep ui-gateway
+        
+    - name: Test public API exports
+      run: |
+        cd PraisonAI/src/praisonai
+        python -c "from praisonai import run_integrated_gateway; print('✓ run_integrated_gateway imported')"
+        python -c "from praisonai import configure_host; print('✓ configure_host imported')"
+        
+    - name: Integration smoke test
+      env:
+        PRAISONAI_TEST_MODE: "1"
+        PYTHONPATH: ${{ github.workspace }}/PraisonAI/src/praisonai:${{ github.workspace }}/PraisonAI/src/praisonai-agents:${{ github.workspace }}/PraisonAIUI
+      run: |
+        cd PraisonAI/src/praisonai
+        python -c "
+        import sys
+        sys.path.insert(0, '.')
+        from praisonai.integration.host_app import configure_host
+        from praisonai import run_integrated_gateway
+        print('✓ Integration imports successful')
+        
+        # Test configure_host with new parameters
+        try:
+            configure_host(
+                style='dashboard', 
+                context_paths=['AGENTS.md'],
+                title='Test App'
+            )
+            print('✓ configure_host with new parameters works')
+        except Exception as e:
+            print(f'× configure_host failed: {e}')
+            sys.exit(1)
+        "
+        
+  # Optional job with API key for agentic tests (if secret is available)
+  agentic-integration:
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    if: ${{ vars.RUN_AGENTIC_TESTS == 'true' }}
+    needs: cross-repo-integration
+    
+    steps:
+    - name: Checkout PraisonAI
+      uses: actions/checkout@v4
+      with:
+        path: PraisonAI
+        ref: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }}
+        
+    - name: Checkout PraisonAIUI
+      uses: actions/checkout@v4
+      with:
+        repository: MervinPraison/PraisonAIUI
+        path: PraisonAIUI
+        ref: main
+        
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: '3.11'
+        
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        cd PraisonAI/src/praisonai-agents
+        pip install -e .
+        cd ../praisonai
+        pip install -e ".[ui,dev]"
+        cd ../../../PraisonAIUI
+        pip install -e .
+        
+    - name: Run agentic integration test
+      env:
+        OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
+      run: |
+        cd PraisonAI/src/praisonai
+        if [ -n "$OPENAI_API_KEY" ]; then
+          python -m pytest tests/integration/test_aiui_host_agentic.py -v --timeout=600
+        else
+          echo "OPENAI_API_KEY not available, skipping agentic tests"
+        fi

.github/workflows/praisonai-issue-triage.yml

@@ -29,9 +29,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           persist-credentials: false
-        with:
-          persist-credentials: false
-        with:
           fetch-depth: 0
           
       - name: Generate GitHub App Token
@@ -63,10 +60,12 @@ jobs:
           GH_TOKEN: ${{ steps.generate_token.outputs.token }}
           OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
           PRAISONAI_AUTO_APPROVE: "true"
+          ISSUE_NUMBER_INPUT: ${{ inputs.issue_number }}
+          ISSUE_NUMBER_EVENT: ${{ github.event.issue.number }}
         run: |
           if [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
-            export ISSUE_NUMBER="${{ inputs.issue_number }}"
+            export ISSUE_NUMBER="$ISSUE_NUMBER_INPUT"
           else
-            export ISSUE_NUMBER="${{ github.event.issue.number }}"
+            export ISSUE_NUMBER="$ISSUE_NUMBER_EVENT"
           fi
           praisonai github triage --issue $ISSUE_NUMBER --agent-file .github/praisonai-issue-triage.yaml

.github/workflows/praisonai-pr-review.yml

@@ -30,22 +30,23 @@ jobs:
     steps:
       - name: Determine checkout ref
         id: dest
+        env:
+          ISSUE_NUMBER: ${{ github.event.issue.number }}
+          PR_NUMBER_INPUT: ${{ inputs.pr_number }}
+          PR_HEAD_SHA: ${{ github.event.pull_request.head.sha }}
         run: |
           if [ "${{ github.event_name }}" = "issue_comment" ]; then
-            echo "ref=refs/pull/${{ github.event.issue.number }}/head" >> "$GITHUB_OUTPUT"
+            echo "ref=refs/pull/$ISSUE_NUMBER/head" >> "$GITHUB_OUTPUT"
           elif [ "${{ github.event_name }}" = "workflow_dispatch" ]; then
-            echo "ref=refs/pull/${{ inputs.pr_number }}/head" >> "$GITHUB_OUTPUT"
+            echo "ref=refs/pull/$PR_NUMBER_INPUT/head" >> "$GITHUB_OUTPUT"
           else
-            echo "ref=${{ github.event.pull_request.head.sha }}" >> "$GITHUB_OUTPUT"
+            echo "ref=$PR_HEAD_SHA" >> "$GITHUB_OUTPUT"
           fi
 
       - name: Checkout Repository
         uses: actions/checkout@v4
         with:
           persist-credentials: false
-        with:
-          persist-credentials: false
-        with:
           ref: ${{ steps.dest.outputs.ref }}
           fetch-depth: 0
           

.github/workflows/test-windows.yml

@@ -26,8 +26,8 @@ jobs:
     
     steps:
     - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
+      with:
+        persist-credentials: false
     - name: Set up Python 3.11
       uses: actions/setup-python@v5
       with:
@@ -176,6 +176,18 @@ jobs:
       run: praisonai queue list
       continue-on-error: true
 
+    # ==================== PACKAGING TESTS ====================
+    - name: Test packaging/daemon entry points
+      run: |
+        echo "Testing both entry point methods..."
+        python -m praisonai --version
+        praisonai --version
+
+    - name: Test doctor packaging command
+      run: |
+        echo "Testing doctor packaging diagnostics..."
+        praisonai doctor packaging --json
+
     # ==================== REAL API TESTS ====================
     - name: Real API test - Agent run
       if: env.OPENAI_API_KEY != ''

README.md

@@ -294,8 +294,35 @@ def search(query: str) -> str:
 
 @tool
 def calculate(expression: str) -> float:
-    """Evaluate a math expression."""
-    return eval(expression)
+    """Safely evaluate a numeric arithmetic expression."""
+    import ast
+    import operator
+    
+    # Define allowed operations
+    _OPS = {
+        ast.Add: operator.add,
+        ast.Sub: operator.sub,
+        ast.Mult: operator.mul,
+        ast.Div: operator.truediv,
+        ast.Pow: operator.pow,
+        ast.USub: operator.neg,
+        ast.UAdd: operator.pos,
+    }
+    
+    def _safe_eval(node):
+        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
+            return node.value
+        elif isinstance(node, ast.BinOp) and type(node.op) in _OPS:
+            return _OPS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))
+        elif isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
+            return _OPS[type(node.op)](_safe_eval(node.operand))
+        else:
+            raise ValueError("Unsupported expression")
+    
+    try:
+        return _safe_eval(ast.parse(expression, mode="eval").body)
+    except (ValueError, SyntaxError, TypeError, ZeroDivisionError, OverflowError):
+        raise ValueError("Invalid arithmetic expression")
 
 agent = Agent(
     instructions="You are a helpful assistant",
@@ -304,6 +331,7 @@ agent = Agent(
 agent.start("Search for AI news and calculate 15*4")
 ```
 
+> ⚠️ **Security Note:** Never use `eval()`, `exec()`, or `subprocess` in tool functions that process LLM-generated or user-supplied input. Always validate and sanitize inputs to prevent code injection attacks.
 > 📖 [Full tools docs](https://docs.praison.ai/docs/tools/tools) — BaseTool, tool packages, 100+ built-in tools
 
 ### 5. Persistence (Databases)
@@ -330,6 +358,15 @@ pip install "praisonai[claw]"
 praisonai claw
 ```
 
+#### Required Environment Variables
+
+Copy `.env.example` to `.env` and configure the following variables:
+
+| Variable | Required | Description |
+|----------|----------|-------------|
+| `OPENAI_API_KEY` | Yes | OpenAI API key for all LLM calls |
+| `TAVILY_API_KEY` | Yes (Claw) | Tavily key for the built-in web-search tool. Get one free at https://app.tavily.com |
+
 Open **http://localhost:8082** — the dashboard comes with 13 built-in pages: Chat, Agents, Memory, Knowledge, Channels, Guardrails, Cron, and more. Add messaging channels directly from the UI.
 
 > 📖 [Full Claw docs](https://docs.praison.ai/docs/concepts/claw) — platform tokens, CLI options, Docker, and YAML agent mode
@@ -345,7 +382,7 @@ praisonai flow
 
 Open **http://localhost:7861** — use the **Agent** and **Agent Team** components to create sequential or parallel workflows. Connect Chat Input → Agent Team → Chat Output for instant multi-agent pipelines.
 
-> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/flow) — visual agent building, component reference, and deployment
+> 📖 [Full Flow docs](https://docs.praison.ai/docs/concepts/agentflow) — visual agent building, component reference, and deployment
 
 ### 8. PraisonAI UI 🤖 (Clean Chat)
 

api.md

@@ -15,6 +15,7 @@ Methods:
 * <code title="class GuardrailResult">GuardrailResult.<a href="./src/praisonai-agents/praisonaiagents/guardrails/guardrail_result.py">from_tuple</a>(result: Tuple[bool, Any]) -> 'GuardrailResult'</code>
 * <code title="class HandoffConfig">HandoffConfig.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">from_dict</a>(data: Dict[str, Any]) -> 'HandoffConfig'</code>
 * <code title="class HandoffConfig">HandoffConfig.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">to_dict</a>() -> Dict[str, Any]</code>
+* <code title="class HandoffResult">HandoffResult.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">from_outcome</a>(outcome: AgentRunOutcome, target_agent: Optional[str] = None, source_agent: Optional[str] = None, handoff_depth: int = 0) -> 'HandoffResult'</code>
 * <code title="class TaskOutput">TaskOutput.<a href="./src/praisonai-agents/praisonaiagents/main.py">json</a>() -> Optional[str]</code>
 * <code title="class TaskOutput">TaskOutput.<a href="./src/praisonai-agents/praisonaiagents/main.py">to_dict</a>() -> dict</code>
 * <code title="class ToolResult">ToolResult.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">to_dict</a>() -> Dict[str, Any]</code>
@@ -40,6 +41,7 @@ Methods:
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">chat_with_context</a>(message: str, context: 'ContextPack', **kwargs) -> str</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">checkpoints</a>() -> Optional[bool]</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">checkpoints</a>(value: Optional[bool]) -> None</code>
+* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">clone_for_channel</a>() -> 'Agent'</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">close</a>() -> None</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">console</a>() -> Optional[Any]</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">context_manager</a>() -> Optional[Any]</code>
@@ -74,7 +76,7 @@ Methods:
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">run_until</a>(prompt: str, criteria: str, threshold: float = 8.0, max_iterations: int = 5, mode: str = 'optimize', on_iteration: Optional[Callable[[Any], None]] = None, verbose: bool = False) -> 'EvaluationLoopResult'</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">run_until_async</a>(prompt: str, criteria: str, threshold: float = 8.0, max_iterations: int = 5, mode: str = 'optimize', on_iteration: Optional[Callable[[Any], None]] = None, verbose: bool = False) -> 'EvaluationLoopResult'</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">skill_manager</a>() -> Optional[Any]</code>
-* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">store_memory</a>(content: str, memory_type: str = 'short_term', **kwargs: Any) -> None</code>
+* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">store_memory</a>(content: str, memory_type: str = 'short_term', action: str = 'add', **kwargs: Any) -> None</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">stream_emitter</a>() -> Optional[Any]</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">stream_emitter</a>(value: Optional[Any]) -> None</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">thinking_budget</a>() -> Optional[int]</code>
@@ -171,8 +173,10 @@ Methods:
 * <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">safe_run</a>(**kwargs) -> ToolResult</code>
 * <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate</a>() -> bool</code>
 * <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate_class</a>() -> bool</code>
+* <code title="class BaseTool">BaseTool.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate_schema_roundtrip</a>() -> bool</code>
 * <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">__call__</a>(*args, **kwargs) -> Any</code>
 * <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">check_availability</a>() -> tuple[bool, str]</code>
+* <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">get_schema</a>() -> Dict[str, Any]</code>
 * <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">injected_params</a>() -> Dict[str, Any]</code>
 * <code title="class FunctionTool">FunctionTool.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">run</a>(**kwargs) -> Any</code>
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">clear</a>() -> None</code>
@@ -180,15 +184,18 @@ Methods:
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">discover_single_file_plugins</a>() -> int</code>
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">get</a>(name: str) -> Optional[Union[BaseTool, Callable]]</code>
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">get_all</a>() -> Dict[str, Union[BaseTool, Callable]]</code>
-* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_available_tools</a>(context: Optional[Dict[str, Any]] = None) -> List[Union[BaseTool, Callable]]</code>
+* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">get_tool_definitions</a>() -> List[Dict[str, Any]]</code>
+* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">get_trust_level</a>(name: str) -> Optional[str]</code>
+* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_available_tools</a>(context: Optional[Dict[str, Any]] = None, ttl_seconds: Optional[float] = None) -> List[Union[BaseTool, Callable]]</code>
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_base_tools</a>() -> List[BaseTool]</code>
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_tools</a>() -> List[str]</code>
-* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">register</a>(tool: Union[BaseTool, Callable], name: Optional[str] = None, overwrite: bool = False) -> None</code>
+* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">list_tools_with_allowed_filter</a>(context: Optional[Dict[str, Any]] = None) -> List[str]</code>
+* <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">register</a>(tool: Union[BaseTool, Callable], name: Optional[str] = None, overwrite: bool = False, trust_level: Optional[str] = None, dynamic_schema_overrides: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None) -> None</code>
 * <code title="class ToolRegistry">ToolRegistry.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">unregister</a>(name: str) -> bool</code>
 * <code title="class Tools">Tools.<a href="./src/praisonai-agents/praisonaiagents/tools/tools.py">internet_search</a>(*args, **kwargs)</code>
 * <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">get_registry</a>() -> ToolRegistry</code>
 * <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">get_tool</a>(name: str) -> Optional[Union[BaseTool, Callable]]</code>
-* <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">register_tool</a>(tool: Union[BaseTool, Callable], name: Optional[str] = None) -> None</code>
+* <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/tools/registry.py">register_tool</a>(tool: Union[BaseTool, Callable], name: Optional[str] = None, trust_level: Optional[str] = None, dynamic_schema_overrides: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None) -> None</code>
 * <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/tools/decorator.py">tool</a>(func: Optional[Callable] = None) -> Union[FunctionTool, Callable[[Callable], FunctionTool]]</code>
 * <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/tools/base.py">validate_tool</a>(tool: Any) -> bool</code>
 
@@ -350,7 +357,7 @@ Methods:
 * <code title="class Handoff">Handoff.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">to_tool_function</a>(source_agent: 'Agent') -> Callable</code>
 * <code title="class Handoff">Handoff.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">tool_description</a>() -> str</code>
 * <code title="class Handoff">Handoff.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">tool_name</a>() -> str</code>
-* <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">handoff</a>(agent: 'Agent', tool_name_override: Optional[str] = None, tool_description_override: Optional[str] = None, on_handoff: Optional[Callable] = None, input_type: Optional[type] = None, input_filter: Optional[Callable[[HandoffInputData], HandoffInputData]] = None, config: Optional[HandoffConfig] = None, context_policy: Optional[str] = None, timeout_seconds: Optional[float] = None, max_concurrent: Optional[int] = None, detect_cycles: Optional[bool] = None, max_depth: Optional[int] = None) -> Handoff</code>
+* <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">handoff</a>(agent: 'Agent', tool_name_override: Optional[str] = None, tool_description_override: Optional[str] = None, on_handoff: Optional[Callable] = None, input_type: Optional[type] = None, input_filter: Optional[Callable[[HandoffInputData], HandoffInputData]] = None, config: Optional[HandoffConfig] = None, context_policy: Optional[str] = None, timeout_seconds: Optional[float] = None, max_concurrent: Optional[int] = None, detect_cycles: Optional[bool] = None, max_depth: Optional[int] = None, tool_policy_mode: Optional[Literal['intersect', 'passthrough']] = None, blocked_tools: Optional[List[str]] = None) -> Handoff</code>
 * <code title="class handoff_filters">handoff_filters.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">compress_history</a>(data: HandoffInputData) -> HandoffInputData</code>
 * <code title="class handoff_filters">handoff_filters.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">keep_last_n_messages</a>(n: int) -> Callable[[HandoffInputData], HandoffInputData]</code>
 * <code title="class handoff_filters">handoff_filters.<a href="./src/praisonai-agents/praisonaiagents/agent/handoff.py">remove_all_tools</a>(data: HandoffInputData) -> HandoffInputData</code>
@@ -476,8 +483,10 @@ Methods:
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">edit_skill</a>(name: str, content: str) -> dict</code>
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_allowed_tools</a>(name: str) -> List[str]</code>
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_available_skills</a>() -> List[SkillMetadata]</code>
+* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_available_skills_by_state</a>(state: SkillState) -> List[LoadedSkill]</code>
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_instructions</a>(name: str) -> Optional[str]</code>
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_skill</a>(name: str) -> Optional[LoadedSkill]</code>
+* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_skills_diagnostics</a>() -> Dict[str, ValidationResult]</code>
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">get_user_invocable_skills</a>() -> List[LoadedSkill]</code>
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">invoke</a>(name: str, raw_args: str = '', session_id: Optional[str] = None, shell_exec: bool = False) -> Optional[str]</code>
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">load_resources</a>(name: str) -> bool</code>
@@ -486,6 +495,7 @@ Methods:
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">skill_names</a>() -> List[str]</code>
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">skills</a>() -> List[LoadedSkill]</code>
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">to_prompt</a>() -> str</code>
+* <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">validate_skill_capabilities</a>(skill_name: str, force_refresh: bool = False) -> ValidationResult</code>
 * <code title="class SkillManager">SkillManager.<a href="./src/praisonai-agents/praisonaiagents/skills/manager.py">write_skill_file</a>(name: str, file_path: str, file_content: str) -> dict</code>
 * <code title="class SkillMetadata">SkillMetadata.<a href="./src/praisonai-agents/praisonaiagents/skills/models.py">from_properties</a>(props: SkillProperties) -> 'SkillMetadata'</code>
 * <code title="class SkillProperties">SkillProperties.<a href="./src/praisonai-agents/praisonaiagents/skills/models.py">to_dict</a>() -> dict</code>
@@ -601,6 +611,13 @@ Methods:
 * <code title="class ContextManager">ContextManager.<a href="./src/praisonai-agents/praisonaiagents/context/manager.py">set_tool_budget</a>(tool_name: str, max_tokens: int, protected: bool = False) -> None</code>
 * <code title="class ContextManager">ContextManager.<a href="./src/praisonai-agents/praisonaiagents/context/manager.py">truncate_tool_output</a>(tool_name: str, output: str) -> str</code>
 
+# UI
+
+Types:
+```python
+from praisonaiagents import A2UI
+```
+
 # Config
 
 Types:
@@ -613,6 +630,7 @@ Methods:
 * <code title="class AutonomyConfig">AutonomyConfig.<a href="./src/praisonai-agents/praisonaiagents/agent/autonomy.py">effective_track_changes</a>() -> bool</code>
 * <code title="class AutonomyConfig">AutonomyConfig.<a href="./src/praisonai-agents/praisonaiagents/agent/autonomy.py">from_dict</a>(data: Dict[str, Any]) -> 'AutonomyConfig'</code>
 * <code title="class CachingConfig">CachingConfig.<a href="./src/praisonai-agents/praisonaiagents/config/feature_configs.py">to_dict</a>() -> Dict[str, Any]</code>
+* <code title="class ExecutionConfig">ExecutionConfig.<a href="./src/praisonai-agents/praisonaiagents/config/feature_configs.py">from_dict</a>(data: Dict[str, Any]) -> 'ExecutionConfig'</code>
 * <code title="class ExecutionConfig">ExecutionConfig.<a href="./src/praisonai-agents/praisonaiagents/config/feature_configs.py">to_dict</a>() -> Dict[str, Any]</code>
 * <code title="class GuardrailConfig">GuardrailConfig.<a href="./src/praisonai-agents/praisonaiagents/config/feature_configs.py">to_dict</a>() -> Dict[str, Any]</code>
 * <code title="class HooksConfig">HooksConfig.<a href="./src/praisonai-agents/praisonaiagents/config/feature_configs.py">to_dict</a>() -> Dict[str, Any]</code>
@@ -705,12 +723,16 @@ Methods:
 * <code title="class AgentFlow">AgentFlow.<a href="./src/praisonai-agents/praisonaiagents/workflows/workflows.py">verbose</a>(value: bool)</code>
 * <code title="class AgentOSProtocol">AgentOSProtocol.<a href="./src/praisonai-agents/praisonaiagents/app/protocols.py">get_app</a>() -> Any</code>
 * <code title="class AgentOSProtocol">AgentOSProtocol.<a href="./src/praisonai-agents/praisonaiagents/app/protocols.py">serve</a>(host: Optional[str] = None, port: Optional[int] = None, reload: bool = False, **kwargs: Any) -> None</code>
+* <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">aannounce_completion</a>(agent_id: str, task_id: str, result: Any, success: bool = True, error: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> None</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">add_task</a>(task)</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">aexecute_task</a>(task_id)</code>
+* <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">announce_completion</a>(agent_id: str, task_id: str, result: Any, success: bool = True, error: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None) -> None</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">append_to_state</a>(key: str, value: Any, max_length: Optional[int] = None) -> List[Any]</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">arun_all_tasks</a>()</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">arun_task</a>(task_id)</code>
+* <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">aspawn_sub_agent</a>(agent: Agent, task: Any, completion_callback: Optional[Callable[[SubAgentCompletionEvent], Any]] = None, metadata: Optional[Dict[str, Any]] = None) -> SpawnedSubAgent</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">astart</a>(content = None, return_dict = False, **kwargs)</code>
+* <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">await_for_completions</a>(timeout: Optional[float] = None, agent_ids: Optional[List[str]] = None) -> List[SubAgentCompletionEvent]</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">clean_json_output</a>(output: str) -> str</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">clear_state</a>() -> None</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">close</a>() -> None</code>
@@ -725,6 +747,7 @@ Methods:
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">get_all_tasks_status</a>()</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">get_detailed_token_report</a>() -> Dict[str, Any]</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">get_plan_markdown</a>() -> str</code>
+* <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">get_spawned_agents</a>() -> List[SpawnedSubAgent]</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">get_state</a>(key: str, default: Any = None) -> Any</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">get_task_details</a>(task_id)</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">get_task_result</a>(task_id)</code>
@@ -741,10 +764,12 @@ Methods:
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">save_output_to_file</a>(task, task_output)</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">save_session_state</a>(session_id: str, include_memory: bool = True) -> None</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">set_state</a>(key: str, value: Any) -> None</code>
+* <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">spawn_sub_agent</a>(agent: Agent, task: Any, completion_callback: Optional[Callable[[SubAgentCompletionEvent], Any]] = None, metadata: Optional[Dict[str, Any]] = None) -> SpawnedSubAgent</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">start</a>(content = None, return_dict = False, output = None, **kwargs)</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">todo_list</a>()</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">update_plan_step_status</a>(step_id: str, status: str) -> bool</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">update_state</a>(updates: Dict) -> None</code>
+* <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">wait_for_completions</a>(timeout: Optional[float] = None, agent_ids: Optional[List[str]] = None) -> List[SubAgentCompletionEvent]</code>
 * <code title="class AutoApproveBackend">AutoApproveBackend.<a href="./src/praisonai-agents/praisonaiagents/approval/backends.py">request_approval</a>(request: ApprovalRequest) -> ApprovalDecision</code>
 * <code title="class AutoApproveBackend">AutoApproveBackend.<a href="./src/praisonai-agents/praisonaiagents/approval/backends.py">request_approval_sync</a>(request: ApprovalRequest) -> ApprovalDecision</code>
 * <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/embedding/embed.py">aembedding</a>(input: Union[str, List[str]], model: str = 'text-embedding-3-small', dimensions: Optional[int] = None, encoding_format: str = 'float', timeout: float = 600.0, api_key: Optional[str] = None, api_base: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs) -> EmbeddingResult</code>
@@ -755,7 +780,7 @@ Methods:
 
 Types:
 ```python
-from praisonai import Agent, AgentApp, AgentOS, AnthropicManagedAgent, CloudProvider, DB, Deploy, DeployConfig, DeployType, HostedAgent, HostedAgentConfig, LocalAgent, LocalAgentConfig, LocalManagedAgent, LocalManagedConfig, ManagedAgent, ManagedConfig, PraisonAI, __version__
+from praisonai import Agent, AgentApp, AgentOS, AnthropicManagedAgent, CloudProvider, DB, Deploy, DeployConfig, DeployType, HostedAgent, HostedAgentConfig, LocalAgent, LocalAgentConfig, LocalManagedAgent, LocalManagedConfig, ManagedAgent, ManagedConfig, PraisonAI, __version__, arun, run, run_integrated_gateway
 ```
 
 # CLI
@@ -787,6 +812,9 @@ Methods:
 * <code title="cli">praisonai bot email <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
 * <code title="cli">praisonai bot install-daemon <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
 * <code title="cli">praisonai bot linear <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
+* <code title="cli">praisonai bot list <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
+* <code title="cli">praisonai bot purge <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
+* <code title="cli">praisonai bot replay <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
 * <code title="cli">praisonai bot slack <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
 * <code title="cli">praisonai bot start <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
 * <code title="cli">praisonai bot telegram <a href="./src/praisonai/praisonai/cli/commands/bot.py">--help</a></code>
@@ -854,8 +882,10 @@ Methods:
 * <code title="cli">praisonai doctor env <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
 * <code title="cli">praisonai doctor mcp <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
 * <code title="cli">praisonai doctor network <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
+* <code title="cli">praisonai doctor packaging <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
 * <code title="cli">praisonai doctor performance <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
 * <code title="cli">praisonai doctor selftest <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
+* <code title="cli">praisonai doctor skills <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
 * <code title="cli">praisonai doctor tools <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
 * <code title="cli">praisonai doctor troubleshoot <a href="./src/praisonai/praisonai/cli/commands/doctor.py">--help</a></code>
 * <code title="cli">praisonai endpoints list <a href="./src/praisonai/praisonai/cli/commands/endpoints.py">--help</a></code>
@@ -887,11 +917,24 @@ Methods:
 * <code title="cli">praisonai gateway send <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
 * <code title="cli">praisonai gateway start <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
 * <code title="cli">praisonai gateway status <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
+* <code title="cli">praisonai gateway stop <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
 * <code title="cli">praisonai gateway uninstall <a href="./src/praisonai/praisonai/cli/commands/gateway.py">--help</a></code>
 * <code title="cli">praisonai github triage <a href="./src/praisonai/praisonai/cli/commands/github.py">--help</a></code>
 * <code title="cli">praisonai hooks add <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
 * <code title="cli">praisonai hooks list <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
 * <code title="cli">praisonai hooks remove <a href="./src/praisonai/praisonai/cli/commands/hooks.py">--help</a></code>
+* <code title="cli">praisonai kanban block <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban boards <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban comment <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban complete <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban create <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban dispatch <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban link <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban list <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban move <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban reclaim <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban show <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
+* <code title="cli">praisonai kanban unblock <a href="./src/praisonai/praisonai/cli/commands/kanban.py">--help</a></code>
 * <code title="cli">praisonai knowledge add <a href="./src/praisonai/praisonai/cli/commands/knowledge.py">--help</a></code>
 * <code title="cli">praisonai knowledge index <a href="./src/praisonai/praisonai/cli/commands/knowledge.py">--help</a></code>
 * <code title="cli">praisonai knowledge list <a href="./src/praisonai/praisonai/cli/commands/knowledge.py">--help</a></code>
@@ -1049,6 +1092,7 @@ Methods:
 * <code title="cli">praisonai serve status <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
 * <code title="cli">praisonai serve stop <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
 * <code title="cli">praisonai serve ui <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
+* <code title="cli">praisonai serve ui-gateway <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
 * <code title="cli">praisonai serve unified <a href="./src/praisonai/praisonai/cli/commands/serve.py">--help</a></code>
 * <code title="cli">praisonai session delete <a href="./src/praisonai/praisonai/cli/commands/session.py">--help</a></code>
 * <code title="cli">praisonai session export <a href="./src/praisonai/praisonai/cli/commands/session.py">--help</a></code>

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>=4.6.37" \
+    "praisonai>=4.6.55" \
     "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>=4.6.37" \
+    "praisonai>=4.6.55" \
     "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>=4.6.37" \
+    "praisonai>=4.6.55" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

examples/benchmark/benchmark_example.py

@@ -35,7 +35,7 @@ def run_quick_comparison():
         prompt="What is 2+2?",
         iterations=2,
         paths=["openai_sdk", "praisonai_agent"],
-        output="verbose"
+        verbose=True
     )
     
     # Print comparison table
@@ -56,7 +56,7 @@ def run_full_benchmark():
     report = handler.run_full_benchmark(
         prompt="Hi",
         iterations=3,
-        output="verbose"
+        verbose=True
     )
     
     # Print full report
@@ -78,7 +78,7 @@ def run_agent_benchmark():
         prompt="Explain Python in one sentence",
         iterations=3,
         paths=["openai_sdk", "praisonai_agent"],
-        output="verbose"
+        verbose=True
     )
     
     # Show timeline diagrams

examples/compare/cli_compare_advanced.py

@@ -22,7 +22,7 @@ custom_modes = {
 }
 
 # Create handler with verbose output
-handler = CompareHandler(output="verbose")
+handler = CompareHandler(verbose=True)
 
 # Example: Manually create comparison results for analysis
 # (useful for testing without API calls)

examples/compare/cli_compare_basic.py

@@ -22,7 +22,7 @@ modes = parse_modes("basic,tools,planning")
 print(f"\nParsed modes: {modes}")
 
 # Create compare handler
-handler = CompareHandler(output="verbose")
+handler = CompareHandler(verbose=True)
 
 # Compare modes (this will run actual agent calls)
 # Uncomment to run:

examples/consolidated_params/advanced_memory.py

@@ -15,7 +15,7 @@ agent = Agent(
 # With array [preset, overrides]
 agent_override = Agent(
     instructions="You are a helpful assistant.",
-    memory=["redis", {"user_id": "custom_user"}],
+    memory=MemoryConfig(backend="redis", user_id="custom_user"),
 )
 
 if __name__ == "__main__":

examples/consolidated_params/basic_workflow.py

@@ -15,8 +15,20 @@ workflow = AgentFlow(
     ],
     output="verbose",
     planning=True,
+    variables={"topic": "AI agents"},
 )
 
 if __name__ == "__main__":
-    result = workflow.run(variables={"topic": "AI agents"})
+    import os
+    import sys
+
+    if not os.environ.get("OPENAI_API_KEY"):
+        print("Set OPENAI_API_KEY to run this example.")
+        sys.exit(0)
+    if sys.platform == "win32":
+        try:
+            sys.stdout.reconfigure(encoding="utf-8")
+        except Exception:
+            pass
+    result = workflow.run("AI agents")
     print(result)

examples/eval/accuracy_example.py

@@ -24,7 +24,7 @@ if has_api_key:
         input_text="What is 2 + 2?",
         expected_output="4",
         num_iterations=3,  # Run 3 times for statistical significance
-        output="verbose"
+        verbose=True
     )
 
     # Run evaluation
@@ -42,7 +42,7 @@ else:
         input_text="What is 2 + 2?",
         expected_output="4",
         num_iterations=1,
-        output="verbose"
+        verbose=True
     )
     
     # Test with mock output

examples/eval/criteria_example.py

@@ -27,7 +27,7 @@ if has_api_key:
         scoring_type="numeric",  # Score 1-10
         threshold=7.0,           # Pass if score >= 7
         num_iterations=2,
-        output="verbose"
+        verbose=True
     )
 
     # Run evaluation
@@ -45,7 +45,7 @@ if has_api_key:
         agent=agent,
         input_text="Tell me a joke",
         scoring_type="binary",  # Pass or Fail
-        output="verbose"
+        verbose=True
     )
 
     binary_result = binary_evaluator.run(print_summary=True)

examples/eval/performance_example.py

@@ -26,7 +26,7 @@ if has_api_key:
         num_iterations=10,  # Run 10 benchmark iterations
         warmup_runs=2,      # 2 warmup runs before measurement
         track_memory=True,  # Track memory usage
-        output="verbose"
+        verbose=True
     )
 
     # Run evaluation

examples/eval/reliability_example.py

@@ -16,8 +16,19 @@ def search_web(query: str) -> str:
 
 
 def calculate(expression: str) -> str:
-    """Calculate a math expression."""
-    return str(eval(expression))
+    """Calculate a math expression safely."""
+    import ast, operator
+    _OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
+            ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
+            ast.Mod: operator.mod, ast.Pow: operator.pow,
+            ast.USub: operator.neg, ast.UAdd: operator.pos}
+    def _ev(n):
+        if isinstance(n, ast.Expression): return _ev(n.body)
+        if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
+        if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
+        if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
+        raise ValueError(f"Unsupported: {ast.dump(n)}")
+    return str(_ev(ast.parse(expression, mode="eval")))
 
 
 # Check if we have an API key
@@ -37,7 +48,7 @@ if has_api_key:
         input_text="Search for the weather in Paris and calculate 25 * 4",
         expected_tools=["search_web", "calculate"],  # Tools that should be called
         forbidden_tools=["delete_file"],  # Tools that should NOT be called
-        output="verbose"
+        verbose=True
     )
 
     # Run evaluation
@@ -59,7 +70,7 @@ evaluator = ReliabilityEvaluator(
     agent=mock_agent,
     expected_tools=["search_web", "calculate"],  # Tools that should be called
     forbidden_tools=["delete_file"],  # Tools that should NOT be called
-    output="verbose"
+    verbose=True
 )
 
 result2 = evaluator.evaluate_tool_calls(

examples/knowledge/compression_demo.py

@@ -184,7 +184,7 @@ Answer questions based ONLY on the provided knowledge context.
 When asked about codes, provide the EXACT code from the documents.
 Be concise but accurate.""",
             knowledge=[temp_dir],
-            user_id="compression_demo_user",
+            memory={"user_id": "compression_demo_user"},
             output="verbose",
         )
         

examples/knowledge/hierarchical_summaries_demo.py

@@ -164,7 +164,7 @@ Answer questions based ONLY on the provided knowledge context.
 When asked about department codes, provide the EXACT code from the documents.
 Be specific about which department you're referring to.""",
             knowledge=[temp_dir],
-            user_id="hierarchy_demo_user",
+            memory={"user_id": "hierarchy_demo_user"},
             output="verbose",
         )
         

examples/knowledge/large_corpus_strategy_demo.py

@@ -124,7 +124,7 @@ def main():
             # Show strategy selection for different corpus sizes
             print(f"\nStrategy selection by corpus size:")
             for size in [100, 1000, 10000, 50000, 100000]:
-                strategy = select_strategy(corpus_tokens=size)
+                strategy = select_strategy(CorpusStats(file_count=max(1, size // 1000), total_tokens=size))
                 print(f"  {size:>6} tokens -> {strategy.value}")
                 
         except ImportError as e:
@@ -142,7 +142,7 @@ Answer questions based ONLY on the provided knowledge context.
 When asked about verification codes, provide the EXACT code from the documents.
 If information is not in the context, say 'Information not found in documents.'""",
             knowledge=[temp_dir],
-            user_id="strategy_demo_user",
+            memory={"user_id": "strategy_demo_user"},
             output="verbose",  # Use new consolidated param
         )
         

examples/knowledge/mem0_backend_example.py

@@ -65,7 +65,7 @@ def main():
             name="PolicyExpert",
             instructions="You are a company policy expert. Answer questions based ONLY on the provided knowledge. Quote exact values when asked about codes or numbers.",
             knowledge=[temp_dir],  # Add directory with documents
-            user_id="example_user",  # Required for mem0 backend
+            memory={"user_id": "example_user"},
         )
         
         print("\n1. Testing Agent.chat() with context-required question:")

examples/knowledge/scope_identifiers_example.py

@@ -50,7 +50,7 @@ def main():
             name="HR_Assistant_User1",
             instructions="You are an HR assistant.",
             knowledge=[temp_dir],
-            user_id="user_alice",  # Knowledge scoped to Alice
+            memory={"user_id": "user_alice"},
         )
         
         response = user1_agent.chat("What is the policy on remote work?")
@@ -80,7 +80,7 @@ def main():
             name="Personal_HR_Bot",
             instructions="You are a personal HR assistant.",
             knowledge=[temp_dir],
-            user_id="user_bob",
+            memory={"user_id": "user_bob"},
             agent_id="hr_bot_v2",
         )
         

examples/knowledge/scope_isolation_example.py

@@ -49,7 +49,7 @@ Only authorized personnel with code WRENCH-992 may access this project.
             name="AgentA_Ingester",
             instructions="You ingest and store knowledge.",
             knowledge=[temp_dir],
-            user_id="user_alice",  # Scoped to Alice
+            memory={"user_id": "user_alice"},
             output="silent",
         )
         
@@ -103,7 +103,7 @@ Only authorized personnel with code WRENCH-992 may access this project.
             name="AgentB_Querier",
             instructions="Answer questions based on the provided knowledge. If you don't have the information, say so.",
             knowledge=[temp_dir],
-            user_id="user_alice",  # Same scope as Agent A
+            memory={"user_id": "user_alice"},
             output="silent",
         )
         

examples/mcp/multi_server.py

@@ -19,20 +19,43 @@ import os
 
 def calculate(expression: str) -> str:
     """
-    A simple calculator tool that evaluates mathematical expressions.
-    
+    A simple calculator tool that evaluates mathematical expressions safely.
+
     Args:
         expression: A mathematical expression to evaluate (e.g., "2 + 2")
-    
+
     Returns:
         The result of the calculation
     """
+    import ast
+    import operator
+
+    _OPS = {
+        ast.Add: operator.add,
+        ast.Sub: operator.sub,
+        ast.Mult: operator.mul,
+        ast.Div: operator.truediv,
+        ast.FloorDiv: operator.floordiv,
+        ast.Mod: operator.mod,
+        ast.Pow: operator.pow,
+        ast.USub: operator.neg,
+        ast.UAdd: operator.pos,
+    }
+
+    def _safe_eval(node):
+        if isinstance(node, ast.Expression):
+            return _safe_eval(node.body)
+        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
+            return node.value
+        if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
+            return _OPS[type(node.op)](_safe_eval(node.operand))
+        if isinstance(node, ast.BinOp) and type(node.op) in _OPS:
+            return _OPS[type(node.op)](_safe_eval(node.left), _safe_eval(node.right))
+        raise ValueError(f"Unsupported expression: {ast.dump(node)}")
+
     try:
-        # Safe evaluation of mathematical expressions
-        allowed_chars = set("0123456789+-*/.() ")
-        if not all(c in allowed_chars for c in expression):
-            return "Error: Invalid characters in expression"
-        result = eval(expression)
+        tree = ast.parse(expression, mode="eval")
+        result = _safe_eval(tree)
         return f"Result: {result}"
     except Exception as e:
         return f"Error: {str(e)}"

examples/mcp_server/custom_tools_server.py

@@ -24,12 +24,18 @@ def main():
     def calculate(expression: str) -> str:
         """Safely evaluate a mathematical expression."""
         try:
-            # Only allow safe math operations
-            allowed = set("0123456789+-*/.(). ")
-            if not all(c in allowed for c in expression):
-                return "Error: Invalid characters in expression"
-            result = eval(expression)  # Safe due to character filtering
-            return f"Result: {result}"
+            import ast, operator
+            _OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
+                    ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
+                    ast.Mod: operator.mod, ast.Pow: operator.pow,
+                    ast.USub: operator.neg, ast.UAdd: operator.pos}
+            def _ev(n):
+                if isinstance(n, ast.Expression): return _ev(n.body)
+                if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
+                if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
+                if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
+                raise ValueError(f"Unsupported: {ast.dump(n)}")
+            return f"Result: {_ev(ast.parse(expression, mode='eval'))}"
         except Exception as e:
             return f"Error: {e}"
     

examples/persistence/knowledge_qdrant.py

@@ -50,10 +50,10 @@ print("=== Qdrant Knowledge Store Demo (Agent-First) ===")
 agent = Agent(
     name="KnowledgeAssistant",
     instructions="You are a helpful assistant with access to technical documentation.",
-    knowledge=[doc_path],
     knowledge={
+        "sources": [doc_path],
         "vector_store": "qdrant",
-        "url": "http://localhost:6333"
+        "url": "http://localhost:6333",
     }
 )
 

examples/python/a2a/a2a-server.py

@@ -21,9 +21,36 @@ def search_web(query: str) -> str:
     return f"Search results for: {query}"
 
 def calculate(expression: str) -> str:
-    """Calculate a mathematical expression."""
+    """Calculate a mathematical expression safely."""
+    import ast
+    import operator
+
+    _OPS = {
+        ast.Add: operator.add,
+        ast.Sub: operator.sub,
+        ast.Mult: operator.mul,
+        ast.Div: operator.truediv,
+        ast.FloorDiv: operator.floordiv,
+        ast.Mod: operator.mod,
+        ast.Pow: operator.pow,
+        ast.USub: operator.neg,
+        ast.UAdd: operator.pos,
+    }
+
+    def _eval(node):
+        if isinstance(node, ast.Expression):
+            return _eval(node.body)
+        if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
+            return node.value
+        if isinstance(node, ast.UnaryOp) and type(node.op) in _OPS:
+            return _OPS[type(node.op)](_eval(node.operand))
+        if isinstance(node, ast.BinOp) and type(node.op) in _OPS:
+            return _OPS[type(node.op)](_eval(node.left), _eval(node.right))
+        raise ValueError(f"Unsupported expression: {ast.dump(node)}")
+
     try:
-        return f"Result: {eval(expression)}"
+        tree = ast.parse(expression, mode="eval")
+        return f"Result: {_eval(tree)}"
     except Exception:
         return "Invalid expression"
 

examples/python/a2ui/send_surface.py

@@ -0,0 +1,27 @@
+#!/usr/bin/env python3
+"""Minimal A2UI example — agent + send_a2ui_messages tool.
+
+Install:
+    pip install praisonaiagents[a2ui]
+
+Run:
+    python send_surface.py
+
+For live rendering, use PraisonAIUI with AG-UI (/agui) — see integrate-a2ui-frontend docs.
+"""
+
+from praisonaiagents import Agent
+from praisonaiagents.tools.a2ui_tools import send_a2ui_messages
+
+agent = Agent(
+    name="ui-assistant",
+    instructions=(
+        "When asked for UI, call send_a2ui_messages with A2UI v0.9 JSON "
+        "(createSurface on surface id 'main')."
+    ),
+    tools=[send_a2ui_messages],
+)
+
+if __name__ == "__main__":
+    result = agent.start("Show a simple welcome card on surface main")
+    print(result)

examples/python/cli/git_integration_example.py

@@ -8,7 +8,7 @@ Docs: https://docs.praison.ai/cli/git-integration
 from praisonai.cli.features import GitIntegrationHandler
 
 # Initialize
-handler = GitIntegrationHandler(output="verbose")
+handler = GitIntegrationHandler(verbose=True)
 git = handler.initialize(repo_path=".")
 
 # Check if it's a git repo

examples/python/cli/repo_map_example.py

@@ -8,7 +8,7 @@ Docs: https://docs.praison.ai/cli/repo-map
 from praisonai.cli.features import RepoMapHandler
 
 # Initialize with current directory
-handler = RepoMapHandler(output="verbose")
+handler = RepoMapHandler(verbose=True)
 repo_map = handler.initialize(root=".")
 
 # Get the repository map

examples/python/cli/sandbox_execution_example.py

@@ -9,7 +9,7 @@ from praisonai.cli.features import SandboxExecutorHandler
 from praisonai.cli.features.sandbox_executor import SandboxMode
 
 # Initialize with basic sandbox
-handler = SandboxExecutorHandler(output="verbose")
+handler = SandboxExecutorHandler(verbose=True)
 sandbox = handler.initialize(mode="basic")
 
 print(f"Sandbox mode: {handler.get_mode()}")

examples/python/concepts/chat-with-pdf.py

@@ -25,8 +25,7 @@ pdf_agent = Agent(
     goal="Help users understand and extract information from PDF documents",
     backstory="You are an expert at reading, analyzing, and answering questions about PDF documents. You provide accurate, detailed answers based on the document content.",
     instructions="Read the provided PDF document and answer questions about its content. Be specific and cite relevant sections when possible.",
-    knowledge=["document.pdf"],  # Replace with your PDF file path
-    knowledge=config,
+    knowledge={**config, "sources": ["document.pdf"]},  # Replace with your PDF path
     reflection=True,
     
     

examples/python/concepts/knowledge-agents-task.py

@@ -1,6 +1,9 @@
+from pathlib import Path
+
 from praisonaiagents import Agent, Task, AgentTeam
 import logging
 import os
+import sys
 
 # Configure logging
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
@@ -17,14 +20,18 @@ config = {
     }
 }
 
+_sample_path = Path(__file__).with_name("sample_knowledge.txt")
+if not _sample_path.exists():
+    _sample_path.write_text("Mervin Praison is the creator of PraisonAI.\n", encoding="utf-8")
+
 # Create an agent with knowledge capabilities
 knowledge_agent = Agent(
     name="KnowledgeAgent",
     role="Information Specialist",
     goal="Store and retrieve knowledge efficiently",
     backstory="Expert in managing and utilizing stored knowledge",
-    knowledge=["sample.pdf"],
-    knowledge=config)
+    knowledge={**config, "sources": [str(_sample_path)]},
+)
 
 # Define a task for the agent
 knowledge_task = Task(
@@ -39,8 +46,16 @@ agents = AgentTeam(
     agents=[knowledge_agent],
     tasks=[knowledge_task],
     process="sequential",
-    user_id="user1"
 )
 
-# Start execution
-result = agents.start()
+if __name__ == "__main__":
+    if not os.environ.get("OPENAI_API_KEY"):
+        print("Set OPENAI_API_KEY to run this example.")
+        sys.exit(0)
+    if sys.platform == "win32":
+        try:
+            sys.stdout.reconfigure(encoding="utf-8")
+        except Exception:
+            pass
+    result = agents.start()
+    print(result)

examples/python/concepts/knowledge-reranker-example.py

@@ -138,8 +138,7 @@ def main():
     agent = Agent(
         name="AI Research Assistant",
         instructions="You are an AI research assistant. Use your knowledge base to answer questions about artificial intelligence topics.",
-        knowledge=sample_documents,  # Will use default knowledge config
-        knowledge=rerank_config,  # Use reranking config
+        knowledge={**rerank_config, "sources": sample_documents},
         llm="gpt-4o-mini"
     )
     

examples/python/concepts/rag-agents.py

@@ -32,8 +32,12 @@ rag_task = Task(
 agents = AgentTeam(
     agents=[rag_agent],
     tasks=[rag_task],
-    user_id="user1"
 )
 
-# Start Agents
-agents.start()
+if __name__ == "__main__":
+    import os
+    import sys
+    if not os.environ.get("OPENAI_API_KEY"):
+        print("Set OPENAI_API_KEY to run this example.")
+        sys.exit(0)
+    agents.start()

examples/python/custom_tools/tools_with_latency.py

@@ -49,7 +49,18 @@ def calculate_with_tracking(expression: str) -> str:
     time.sleep(0.05)
     
     try:
-        result = eval(expression)
+        import ast, operator
+        _OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
+                ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
+                ast.Mod: operator.mod, ast.Pow: operator.pow,
+                ast.USub: operator.neg, ast.UAdd: operator.pos}
+        def _ev(n):
+            if isinstance(n, ast.Expression): return _ev(n.body)
+            if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
+            if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
+            if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
+            raise ValueError(f"Unsupported: {ast.dump(n)}")
+        result = _ev(ast.parse(expression, mode="eval"))
         return f"Result: {result}"
     except Exception as e:
         return f"Error: {str(e)}"

examples/python/general/memory_example.py

@@ -5,7 +5,7 @@ import os
 def main():
     # Initialize memory config
     memory_config = {
-        "provider": "rag",
+        "backend": "sqlite",
         "use_embedding": True,
         "storage": {
             "type": "sqlite",
@@ -105,7 +105,7 @@ def main():
         agents=[researcher, retriever],
         tasks=[store_task, verify_task, query_task, query_both_task],  # Use same verbose level as memory
         memory={
-            "provider": "rag",
+            "backend": "sqlite",
             "embedder": {
                 "provider": "openai",
                 "config": {

examples/python/managed-agents/provider/local_advanced.py

@@ -23,12 +23,39 @@ print(f"[2] Updated to v{managed.agent_version}")
 result = agent.start("What is 123 + 456?", stream=True)
 
 # ── 3. Custom tool ──
+def _safe_calc(expr: str) -> str:
+    import ast
+    allowed = set("0123456789+-*/.() ")
+    if not all(c in allowed for c in expr):
+        return "error"
+    try:
+        tree = ast.parse(expr, mode="eval")
+        for node in ast.walk(tree):
+            if not isinstance(
+                node,
+                (
+                    ast.Expression,
+                    ast.BinOp,
+                    ast.UnaryOp,
+                    ast.Constant,
+                    ast.Add,
+                    ast.Sub,
+                    ast.Mult,
+                    ast.Div,
+                    ast.USub,
+                    ast.UAdd,
+                ),
+            ):
+                return "error"
+        val = eval(compile(tree, "<expr>", "eval"), {"__builtins__": {}})
+        return str(val)
+    except Exception:
+        return "error"
+
+
 def handle_calculator(tool_name, tool_input):
     expr = tool_input.get("expression", "0")
-    try:
-        val = eval(expr, {"__builtins__": {}})
-    except Exception:
-        val = "error"
+    val = _safe_calc(str(expr))
     print(f"  [Calculator: {expr} = {val}]")
     return str(val)
 

examples/python/mcp/mcp-tools-server-sse.py

@@ -36,10 +36,18 @@ def calculate(expression: str) -> dict:
         expression: Math expression (e.g., "2 + 2 * 3")
     """
     try:
-        allowed = set("0123456789+-*/.() ")
-        if not all(c in allowed for c in expression):
-            return {"error": "Invalid characters"}
-        return {"expression": expression, "result": eval(expression)}
+        import ast, operator
+        _OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
+                ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
+                ast.Mod: operator.mod, ast.Pow: operator.pow,
+                ast.USub: operator.neg, ast.UAdd: operator.pos}
+        def _ev(n):
+            if isinstance(n, ast.Expression): return _ev(n.body)
+            if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
+            if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
+            if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
+            raise ValueError(f"Unsupported: {ast.dump(n)}")
+        return {"expression": expression, "result": _ev(ast.parse(expression, mode="eval"))}
     except Exception as e:
         return {"error": str(e)}
 

examples/python/mcp/mcp-tools-server.py

@@ -47,12 +47,18 @@ def calculate(expression: str) -> Dict[str, Any]:
         The result of the calculation
     """
     try:
-        # Safe evaluation of mathematical expressions
-        allowed_chars = set("0123456789+-*/.() ")
-        if not all(c in allowed_chars for c in expression):
-            return {"error": "Invalid characters in expression"}
-        
-        result = eval(expression)  # Use safer alternative in production
+        import ast, operator
+        _OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
+                ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
+                ast.Mod: operator.mod, ast.Pow: operator.pow,
+                ast.USub: operator.neg, ast.UAdd: operator.pos}
+        def _ev(n):
+            if isinstance(n, ast.Expression): return _ev(n.body)
+            if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
+            if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
+            if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
+            raise ValueError(f"Unsupported: {ast.dump(n)}")
+        result = _ev(ast.parse(expression, mode="eval"))
         return {"expression": expression, "result": result}
     except Exception as e:
         return {"error": str(e)}

examples/python/mcp_server_example.py

@@ -49,13 +49,18 @@ def calculate(expression: str) -> Dict[str, Any]:
         The result of the calculation
     """
     try:
-        # Safe evaluation of mathematical expressions
-        # In production, use a proper math parser
-        allowed_chars = set("0123456789+-*/.() ")
-        if not all(c in allowed_chars for c in expression):
-            return {"error": "Invalid characters in expression"}
-        
-        result = eval(expression)  # Note: Use safer alternative in production
+        import ast, operator
+        _OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
+                ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
+                ast.Mod: operator.mod, ast.Pow: operator.pow,
+                ast.USub: operator.neg, ast.UAdd: operator.pos}
+        def _ev(n):
+            if isinstance(n, ast.Expression): return _ev(n.body)
+            if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
+            if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
+            if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
+            raise ValueError(f"Unsupported: {ast.dump(n)}")
+        result = _ev(ast.parse(expression, mode="eval"))
         return {"expression": expression, "result": result}
     except Exception as e:
         return {"error": str(e)}

examples/python/memory/advanced-graph-memory-integration.py

@@ -60,8 +60,8 @@ query_task = Task(
 agents_system = AgentTeam(
     agents=[knowledge_agent],
     tasks=[build_task, query_task],
-    memory=True,
-    memory=memory_config, output="verbose"
+    memory=memory_config,
+    output="verbose",
 )
 
 print("Starting graph memory demonstration...")

examples/python/memory/advanced-memory-search-optimization.py

@@ -61,8 +61,8 @@ search_task = Task(
 agents_system = AgentTeam(
     agents=[research_agent],
     tasks=[build_memory_task, search_task],
-    memory=True,
-    memory=memory_config, output="verbose"
+    memory=memory_config,
+    output="verbose",
 )
 
 print("Starting memory search optimization demonstration...")

examples/python/memory/gemini-embedding-example.py

@@ -86,7 +86,7 @@ def main():
         agents=[researcher, retriever],
         tasks=[store_task, retrieve_task],
         memory={
-            "provider": "rag",
+            "backend": "sqlite",
             "embedder": {
                 "provider": "gemini",
                 "config": {

examples/python/memory/graph-memory-agent.py

@@ -152,8 +152,7 @@ def main():
     agents_system = AgentTeam(
         agents=[researcher],
         tasks=[build_knowledge_task, query_knowledge_task, expand_knowledge_task],
-        memory=True,
-        memory=memory_config
+        memory=memory_config,
     )
     
     print("\n🚀 Starting graph memory demonstration...")

examples/python/memory_with_tracing_example.py

@@ -30,7 +30,7 @@ def main():
     try:
         # Create memory instance
         memory = Memory(config={
-            "provider": "rag",
+            "backend": "sqlite",
             "use_embedding": False,
             "agent_name": "demo_agent",  # Used in trace events
         })

examples/python/message_steering_example.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+"""
+Message Steering Example
+
+Demonstrates real-time message steering for agents during execution.
+Shows how to send guidance messages to agents while they're running tasks.
+
+Usage:
+    # Python API
+    python message_steering_example.py
+    
+    # CLI with steering enabled
+    praisonai --message-steering "Write a detailed report about AI trends" 
+    
+    # YAML (add message_steering: true to agent config)
+"""
+
+import asyncio
+import threading
+import time
+from praisonaiagents import Agent
+
+
+def basic_message_steering():
+    """Basic message steering example - enable steering and send guidance."""
+    print("🎯 Basic Message Steering Example")
+    print("=" * 50)
+    
+    # Create agent with message steering enabled
+    agent = Agent(
+        name="researcher", 
+        instructions="You are a helpful research assistant",
+        message_steering=True,  # Enable real-time message steering
+        llm="gpt-4o-mini"
+    )
+    
+    print("✅ Agent created with message steering enabled")
+    print(f"📊 Steering status: {agent.get_steering_status()}")
+    
+    # Send some guidance
+    msg_id_1 = agent.steer("Focus on recent developments from 2024")
+    msg_id_2 = agent.steer("Keep response under 200 words", priority=10)
+    
+    print(f"📨 Sent steering messages: {msg_id_1}, {msg_id_2}")
+    print(f"📊 Steering status: {agent.get_steering_status()}")
+    
+    # Execute task - steering will be automatically injected
+    result = agent.start("Summarize the latest AI trends")
+    
+    print("\n📝 Result:")
+    print(result)
+
+
+def threaded_message_steering():
+    """Advanced example - send steering messages while agent is running."""
+    print("\n🚀 Threaded Message Steering Example")
+    print("=" * 50)
+    
+    agent = Agent(
+        name="writer",
+        instructions="You are a content writer", 
+        message_steering=True,
+        llm="gpt-4o-mini"
+    )
+    
+    result_holder = [None]
+    
+    def run_agent():
+        """Run agent in background thread."""
+        result = agent.start("Write a comprehensive article about quantum computing")
+        result_holder[0] = result
+    
+    # Start agent in background
+    thread = threading.Thread(target=run_agent)
+    thread.start()
+    
+    # Send steering messages while it's running
+    time.sleep(0.5)  # Let execution start
+    agent.steer("Make it accessible to beginners")
+    
+    time.sleep(1.0) 
+    agent.steer("Include practical applications", priority=15)
+    
+    time.sleep(1.5)
+    agent.steer("Keep it under 300 words total", priority=20)
+    
+    # Wait for completion
+    thread.join(timeout=60)
+    
+    if result_holder[0]:
+        print("\n📝 Final Result:")
+        print(result_holder[0])
+    else:
+        print("⚠️ Agent execution timed out")
+
+
+async def async_message_steering():
+    """Async example showing message steering with async execution."""
+    print("\n⚡ Async Message Steering Example")
+    print("=" * 50)
+    
+    agent = Agent(
+        name="analyst",
+        instructions="You are a data analyst",
+        message_steering=True,
+        llm="gpt-4o-mini"
+    )
+    
+    # Start async task
+    async def run_analysis():
+        return await agent.astart("Analyze the impact of AI on job markets")
+    
+    # Create task
+    task = asyncio.create_task(run_analysis())
+    
+    # Send steering while running
+    await asyncio.sleep(0.3)
+    agent.steer("Focus on positive opportunities")
+    
+    await asyncio.sleep(0.7)  
+    agent.steer("Include specific examples", priority=12)
+    
+    # Wait for result
+    result = await task
+    
+    print("\n📝 Analysis Result:")
+    print(result)
+
+
+def cli_yaml_examples():
+    """Show CLI and YAML usage examples."""
+    print("\n💻 CLI & YAML Examples")
+    print("=" * 50)
+    
+    print("🔧 CLI Usage:")
+    print("  praisonai --message-steering 'Your task here'")
+    print("  # Then in another terminal:")
+    print("  # agent.steer('Additional guidance')")
+    
+    print("\n📄 YAML Configuration:")
+    yaml_example = """
+# agents.yaml
+framework: praisonai
+roles:
+  researcher:
+    role: Research Assistant
+    goal: Conduct thorough research
+    backstory: Expert researcher with steering support
+    message_steering: true    # Enable steering for this agent
+    tools:
+      - internet_search
+      - write_file
+    tasks:
+      research_task:
+        description: Research AI trends and write report
+        expected_output: Comprehensive research report
+"""
+    print(yaml_example.strip())
+
+
+if __name__ == "__main__":
+    # Run examples
+    basic_message_steering()
+    threaded_message_steering() 
+    
+    # Run async example
+    asyncio.run(async_message_steering())
+    
+    # Show CLI/YAML usage
+    cli_yaml_examples()
+    
+    print("\n✨ Message steering examples completed!")
+    print("📚 See docs for more advanced patterns and priority levels")

examples/python/models/deepseek/deepseek-rag-agents-streamlit.py

@@ -32,9 +32,8 @@ def init_agent():
     return Agent(
         name="Knowledge Agent",
         instructions="You answer questions based on the provided knowledge.",
-        knowledge=["kag-research-paper.pdf"],
-        knowledge=config,
-        user_id="user1",
+        knowledge={**config, "sources": ["kag-research-paper.pdf"]},
+        memory={"user_id": "user1"},
         llm="deepseek-r1"
     )
 

examples/python/models/deepseek/deepseek-rag-agents.py

@@ -30,9 +30,8 @@ config = {
 agent = Agent(
     name="Knowledge Agent",
     instructions="You answer questions based on the provided knowledge.",
-    knowledge=["kag-research-paper.pdf"], # Indexing
-    knowledge=config,
-    user_id="user1",
+    knowledge={**config, "sources": ["kag-research-paper.pdf"]},
+    memory={"user_id": "user1"},
     llm="deepseek-r1"
 )
 

examples/python/mongodb/mongodb_comprehensive_example.py

@@ -189,8 +189,8 @@ def main():
     business_system = AgentTeam(
         agents=[data_manager, knowledge_curator, business_analyst, customer_service],
         tasks=business_tasks,
-        memory=True,
-        memory=mongodb_memory_config, output="verbose"
+        memory=mongodb_memory_config,
+        output="verbose",
     )
     
     # Execute the comprehensive business pipeline

examples/python/mongodb/mongodb_knowledge_example.py

@@ -55,8 +55,10 @@ def main():
         retrieve information from a comprehensive MongoDB knowledge base. You excel 
         at finding relevant information, synthesizing knowledge from multiple sources, 
         and providing accurate, context-aware responses.""",
-        knowledge=mongodb_knowledge_config,
-        knowledge=[os.path.join(os.path.dirname(__file__), "llms.md")],
+        knowledge={
+            **mongodb_knowledge_config,
+            "sources": [os.path.join(os.path.dirname(__file__), "llms.md")],
+        },
         memory=True,
         llm="gpt-4o-mini"
     )

examples/python/mongodb/mongodb_memory_example.py

@@ -93,8 +93,8 @@ def main():
     research_system = AgentTeam(
         agents=[research_agent],
         tasks=research_tasks,
-        memory=True,
-        memory=mongodb_memory_config, output="verbose"
+        memory=mongodb_memory_config,
+        output="verbose",
     )
     
     # Execute the research pipeline

examples/python/rag/agent_vs_auto_rag.py

@@ -50,7 +50,7 @@ agent = Agent(
     name="TechSupport",
     instructions="You are a technical support assistant.",
     knowledge=[sample_doc],
-    user_id="comparison_user",
+    memory={"user_id": "comparison_user"},
     llm="openai/gpt-4o-mini",
     output="minimal",
 )

examples/python/rag/auto_rag_agent_basic.py

@@ -20,8 +20,9 @@ if not os.environ.get("OPENAI_API_KEY"):
     exit(1)
 
 # Create a sample document for testing
-sample_doc = "/tmp/sample_doc.txt"
-with open(sample_doc, "w") as f:
+from pathlib import Path
+sample_doc = str(Path(__file__).with_name("sample_doc.txt"))
+with open(sample_doc, "w", encoding="utf-8") as f:
     f.write("""
 PraisonAI Framework Documentation
 
@@ -44,7 +45,7 @@ agent = Agent(
     name="DocBot",
     instructions="You are a helpful documentation assistant. Answer questions based on the provided context.",
     knowledge=[sample_doc],
-    user_id="example_user",  # Required for RAG retrieval
+    memory={"user_id": "example_user"},
     llm="openai/gpt-4o-mini",
     output="minimal",
 )

examples/python/rag/auto_rag_agent_policy.py

@@ -38,7 +38,7 @@ agent = Agent(
     name="PolicyBot",
     instructions="You are a customer service assistant.",
     knowledge=[sample_doc],
-    user_id="policy_user",
+    memory={"user_id": "policy_user"},
     llm="openai/gpt-4o-mini",
     output="minimal",
 )

examples/python/rag/rag_basic.py

@@ -44,7 +44,7 @@ agent = Agent(
     name="MLExpert",
     instructions="You are a machine learning expert. Answer questions based on the provided context.",
     knowledge=[sample_doc],
-    user_id="rag_user",
+    memory={"user_id": "rag_user"},
     llm="openai/gpt-4o-mini",
     output="minimal",
 )

examples/python/rag/rag_hybrid.py

@@ -59,7 +59,7 @@ agent = Agent(
     name="APIDocBot",
     instructions="You are an API documentation assistant.",
     knowledge=[sample_doc],
-    user_id="hybrid_user",
+    memory={"user_id": "hybrid_user"},
     llm="openai/gpt-4o-mini",
     output="minimal",
 )

examples/python/rag/rag_with_citations.py

@@ -44,7 +44,7 @@ agent = Agent(
     name="ResearchBot",
     instructions="You are a research assistant. Always cite your sources.",
     knowledge=[doc1, doc2],
-    user_id="citation_user",
+    memory={"user_id": "citation_user"},
     llm="openai/gpt-4o-mini",
     output="minimal",
 )

examples/python/sessions/comprehensive-session-management.py

@@ -172,7 +172,7 @@ print("="*70)
 # Create sessions for different users working on the same project
 reviewer_session = Session(
     session_id="peer_review_001",
-    user_id="user_reviewer_01",  # Different user
+    user_id="user_reviewer_01",
     storage_path=session_dir
 )
 

examples/python/stateful/memory-quality-example.py

@@ -14,13 +14,12 @@ def main():
     
     # Initialize memory with quality-based configuration
     memory_config = {
-        "provider": "rag",
-        "use_embedding": True,
-        "rag_db_path": "quality_memory_db"
+        "backend": "sqlite",
+        "user_id": "quality_demo_user",
     }
     
     memory = Memory(config=memory_config, verbose=5)
-    print(f"📦 Initialized memory system with provider: {memory_config['provider']}")
+    print(f"📦 Initialized memory system with backend: {memory_config['backend']}")
     
     # Create agent with memory
     agent = Agent(
@@ -28,7 +27,6 @@ def main():
         role="Information Quality Specialist",
         instructions="Store and retrieve information with quality assessment",
         memory=memory,
-        user_id="quality_demo_user"
     )
     
     print(f"🤖 Created agent: {agent.name}")
@@ -181,7 +179,7 @@ def main():
     
     context = memory.build_context_for_task(
         task_descr="Explain AI alignment challenges",
-        user_id="quality_demo_user",
+        memory={"user_id": "quality_demo_user"},
         max_items=3
     )
     

examples/python/stateful/session-example.py

@@ -15,7 +15,7 @@ def main():
     # Create a session with persistent storage
     session = Session(
         session_id="demo_chat_001",
-        user_id="demo_user"
+        memory={"user_id": "demo_user"},
     )
     
     print(f"📝 Created session: {session}")

examples/python/stateful/workflow-state-example.py

@@ -88,9 +88,8 @@ def main():
     workflow = AgentTeam(
         agents=[researcher, analyzer, writer],
         tasks=[research_task, analysis_task, report_task],
-        memory=True,
+        memory={"user_id": "research_project_001"},
         process="workflow",
-        user_id="research_project_001",
     )
     
     print(f"🤖 Created workflow with {len(workflow.agents)} agents and {len(workflow.tasks)} tasks")

examples/python/toolsets_example.py

@@ -0,0 +1,227 @@
+"""
+Named Toolsets Example
+
+This example demonstrates how to use named toolsets to organize and reuse 
+groups of tools across agents and workflows.
+
+Toolsets provide:
+- Organized tool collections (web, research, development, etc.)
+- Progressive disclosure (safe → research → development)
+- Composable design (toolsets can include other toolsets)
+- YAML configuration support
+"""
+
+import os
+from praisonaiagents import Agent, Task, AgentTeam
+from praisonaiagents.toolsets import (
+    list_toolsets, 
+    resolve_toolset,
+    register_toolset
+)
+
+# Set up environment (replace with your actual API key)
+os.environ.setdefault("OPENAI_API_KEY", "your-openai-api-key-here")
+
+
+def demo_basic_toolsets():
+    """Demonstrate basic toolset functionality."""
+    print("=== Basic Toolsets Demo ===")
+    
+    # List available toolsets
+    available_toolsets = list_toolsets()
+    print(f"Available toolsets: {available_toolsets}")
+    
+    # Show what tools are in each prebuilt toolset
+    for toolset_name in ["safe", "web", "research"]:
+        tools = resolve_toolset(toolset_name)
+        print(f"\n{toolset_name.upper()} toolset contains: {tools}")
+
+
+def demo_agent_with_toolsets():
+    """Demonstrate agents using toolsets."""
+    print("\n=== Agent with Toolsets Demo ===")
+    
+    # Create a research agent with research toolset
+    researcher = Agent(
+        name="researcher",
+        role="Research Specialist",
+        goal="Research topics thoroughly using web tools",
+        instructions="You are a research specialist with access to web search and file tools.",
+        toolsets=["research"],  # includes web tools + file tools
+        llm="gpt-4o-mini"
+    )
+    
+    print(f"Research agent has {len(researcher.tools)} tools available")
+    print(f"Tools: {[getattr(t, '__name__', str(t)) for t in researcher.tools]}")
+    
+    # Create a safe agent with minimal tools
+    safe_agent = Agent(
+        name="safe_assistant",
+        role="Safe Assistant", 
+        goal="Help users safely without risky operations",
+        instructions="You are a helpful assistant with access to only safe, read-only tools.",
+        toolsets=["safe"],
+        llm="gpt-4o-mini"
+    )
+    
+    print(f"\nSafe agent has {len(safe_agent.tools)} tools available")
+    print(f"Safe tools: {[getattr(t, '__name__', str(t)) for t in safe_agent.tools]}")
+
+
+def demo_custom_toolsets():
+    """Demonstrate creating and using custom toolsets."""
+    print("\n=== Custom Toolsets Demo ===")
+    
+    # Register a custom toolset for data analysis
+    register_toolset(
+        "data_analysis",
+        tools=["read_file", "write_file"],
+        includes=["safe"],  # Include safe tools too
+        description="Tools for data analysis workflows"
+    )
+    
+    # Create an agent with the custom toolset
+    analyst = Agent(
+        name="data_analyst",
+        role="Data Analyst",
+        goal="Analyze data files and generate reports",
+        instructions="You specialize in data analysis. Read files, process data, and write reports.",
+        toolsets=["data_analysis"],
+        llm="gpt-4o-mini"
+    )
+    
+    print(f"Data analyst has {len(analyst.tools)} tools available")
+    print(f"Analysis tools: {[getattr(t, '__name__', str(t)) for t in analyst.tools]}")
+
+
+def demo_mixed_tools_and_toolsets():
+    """Demonstrate mixing explicit tools with toolsets."""
+    print("\n=== Mixed Tools and Toolsets Demo ===")
+    
+    # Create agent with both explicit tools and toolsets
+    hybrid_agent = Agent(
+        name="hybrid_assistant",
+        role="Hybrid Assistant",
+        goal="Assist with both research and custom tasks",
+        instructions="You can research topics and perform custom analysis tasks.",
+        tools=["write_file"],  # Explicit tool
+        toolsets=["research"],  # Toolset with web + file tools
+        llm="gpt-4o-mini"
+    )
+    
+    print(f"Hybrid agent has {len(hybrid_agent.tools)} tools available")
+    tool_names = [getattr(t, '__name__', str(t)) for t in hybrid_agent.tools]
+    print(f"Hybrid tools: {tool_names}")
+    
+    # Note: write_file might appear twice (once explicit, once from toolset)
+    # The agent handles deduplication automatically
+
+
+def demo_multi_agent_with_toolsets():
+    """Demonstrate multi-agent workflows with different toolsets."""
+    print("\n=== Multi-Agent with Toolsets Demo ===")
+    
+    # Create agents with different toolsets for different roles
+    researcher = Agent(
+        name="researcher",
+        role="Research Specialist",
+        goal="Research topics and gather information",
+        toolsets=["research"],
+        llm="gpt-4o-mini"
+    )
+    
+    writer = Agent(
+        name="writer",
+        role="Content Writer",  
+        goal="Write content based on research",
+        toolsets=["safe"],  # Safe tools for writing
+        llm="gpt-4o-mini"
+    )
+    
+    # Create tasks for each agent
+    research_task = Task(
+        name="research_task",
+        description="Research the benefits of renewable energy",
+        expected_output="A summary of key benefits of renewable energy",
+        agent=researcher
+    )
+    
+    writing_task = Task(
+        name="writing_task", 
+        description="Write a blog post about renewable energy based on the research",
+        expected_output="A well-written blog post about renewable energy benefits",
+        agent=writer
+    )
+    
+    # Create team with different toolsets per agent
+    team = AgentTeam(
+        agents=[researcher, writer],
+        tasks=[research_task, writing_task],
+        process="sequential"
+    )
+    
+    print(f"Team created with {len(team.agents)} agents using different toolsets")
+    print(f"Research agent tools: {len(researcher.tools)}")
+    print(f"Writer agent tools: {len(writer.tools)}")
+
+
+def demo_yaml_config():
+    """Show how toolsets work in YAML configuration."""
+    print("\n=== YAML Configuration Example ===")
+    
+    yaml_config = """
+# agents.yaml - Example using toolsets
+agents:
+  researcher:
+    role: Research Specialist
+    goal: Research topics thoroughly
+    toolsets:
+      - research  # Includes web search, file tools
+    llm: gpt-4o-mini
+    
+  analyst:
+    role: Data Analyst
+    goal: Analyze research data
+    tools:
+      - write_file  # Explicit tool
+    toolsets:
+      - safe  # Safe toolset
+    llm: gpt-4o-mini
+
+tasks:
+  research_task:
+    description: Research renewable energy trends
+    agent: researcher
+    expected_output: Research summary
+    
+  analysis_task:
+    description: Analyze the research data
+    agent: analyst
+    expected_output: Analysis report
+"""
+    
+    print("Example YAML configuration with toolsets:")
+    print(yaml_config)
+    print("Use: praisonai run 'Research renewable energy' --toolset research")
+    print("Or:  praisonai chat --toolset web,files")
+
+
+if __name__ == "__main__":
+    print("🛠️  PraisonAI Named Toolsets Example")
+    print("=" * 50)
+    
+    # Run all demos
+    demo_basic_toolsets()
+    demo_agent_with_toolsets()
+    demo_custom_toolsets()
+    demo_mixed_tools_and_toolsets()
+    demo_multi_agent_with_toolsets()
+    demo_yaml_config()
+    
+    print("\n✅ Toolsets example completed!")
+    print("\n💡 Key Benefits:")
+    print("   • Organized tool collections")
+    print("   • Progressive disclosure (safe → research → development)")
+    print("   • Reusable across agents and workflows") 
+    print("   • YAML configuration support")
+    print("   • Composable design with includes")

examples/python/ui/ollama-rag-agents-streamlit.py

@@ -32,9 +32,8 @@ def init_agent():
     return Agent(
         name="Knowledge Agent",
         instructions="You answer questions based on the provided knowledge.",
-        knowledge=["kag-research-paper.pdf"],
-        knowledge=config,
-        user_id="user1",
+        knowledge={**config, "sources": ["kag-research-paper.pdf"]},
+        memory={"user_id": "user1"},
         llm="deepseek-r1"
     )
 

examples/python/usecases/domain-context-solution.py

@@ -264,13 +264,12 @@ agents = AgentTeam(
     tasks=[tool_query_task, pentester_analysis_task, ciso_review_task],
     process="hierarchical",
     manager_llm="gpt-4o-mini",
-    memory=True,  # Enable shared memory for domain context
     memory={
-        "provider": "rag",
+        "backend": "sqlite",
         "config": {
             "collection_name": f"{domain.replace('.', '_')}_security_analysis"
-        }
-    }
+        },
+    },
 )
 
 # Main Execution Workflow

examples/python/workflows/simple_workflow.py

@@ -5,7 +5,7 @@ Demonstrates using Agent instances directly as workflow steps.
 Each agent processes the input and passes output to the next agent.
 """
 
-from praisonaiagents import Agent, Workflow
+from praisonaiagents import Agent, AgentFlow
 
 # Create agents with specific roles
 researcher = Agent(

examples/python/workflows/workflow_conditional.py

@@ -5,7 +5,7 @@ Demonstrates using should_run to conditionally execute agent steps
 based on input or previous results.
 """
 
-from praisonaiagents import Agent, Workflow, Task
+from praisonaiagents import Agent, AgentFlow, Task
 
 # Condition functions
 def is_technical(ctx) -> bool:
@@ -54,11 +54,13 @@ workflow = AgentFlow(
         analyzer,  # Always runs
         Task(
             name="tech_review",
+            description="Review technical aspects of the content",
             agent=tech_reviewer,
             should_run=is_technical  # Only runs for technical content
         ),
         Task(
             name="creative_enhance",
+            description="Enhance the creative aspects of the content",
             agent=creative_enhancer,
             should_run=needs_creative  # Only runs for creative content
         ),

examples/rag/external_sources.py

@@ -207,7 +207,7 @@ def api_augmented_rag():
         instructions="""You are a financial assistant with access to market APIs.
         Provide accurate financial information and context.
         Always note that prices are subject to change.""",
-        user_id="api_demo"
+        memory={"user_id": "api_demo"},
     )
     
     queries = [

examples/scientific_writing/agents.yaml

@@ -0,0 +1,40 @@
+# Scientific paper workflow — YAML equivalent of multi_agent_paper.py
+# Run with:
+#   praisonai agents run --file examples/scientific_writing/agents.yaml
+framework: praisonai
+topic: "Climate change effects on coral reef biodiversity"
+
+roles:
+  literature_reviewer:
+    role: "Literature Reviewer"
+    goal: "Survey recent academic work on {topic} and produce an APA review."
+    backstory: "Expert in academic literature surveying and citation hygiene."
+    llm: "gpt-4o-mini"
+    tasks:
+      review_literature:
+        description: "Produce a concise literature review with 5-8 APA citations."
+        expected_output: "APA-formatted literature review."
+
+  methodology_designer:
+    role: "Methodology Designer"
+    goal: "Design a reproducible research methodology for {topic}."
+    backstory: "Specialises in rigorous experimental design."
+    llm: "gpt-4o-mini"
+    tasks:
+      design_methodology:
+        description: "Write a reproducible methods section in LaTeX."
+        expected_output: "LaTeX methods section."
+
+  scientific_writer:
+    role: "Scientific Writer"
+    goal: "Assemble the final LaTeX paper on {topic}."
+    backstory: >-
+      Specialised in LaTeX-formatted academic papers, fine-tuned on
+      scientific literature (CAJAL-4B checkpoint).
+    llm: "huggingface/Agnuxo/CAJAL-4B-P2PCLAW"
+    tasks:
+      write_paper:
+        description: >-
+          Combine the literature review and methodology into a full
+          scientific paper on {topic}. Use LaTeX formatting.
+        expected_output: "Complete LaTeX paper."

examples/scientific_writing/multi_agent_paper.py

@@ -0,0 +1,73 @@
+"""Multi-agent scientific paper workflow (Researcher -> Methodologist -> Writer).
+
+Demonstrates the agent-centric pattern from AGENTS.md: compose specialised
+Agents with ``AgentTeam`` + sequential ``Task``s instead of creating custom
+Agent subclasses.  The writer uses the CAJAL-4B HuggingFace model; the
+reviewer and methodologist use general-purpose LLMs.
+
+Usage:
+    python examples/scientific_writing/multi_agent_paper.py
+"""
+from praisonaiagents import Agent, AgentTeam, Task
+
+from scientific_writer import (  # noqa: E402 — example imports a sibling file
+    build_scientific_writer,
+    format_citation,
+    format_latex_section,
+)
+
+
+literature_reviewer = Agent(
+    name="Literature Reviewer",
+    instructions=(
+        "You are an expert at surveying academic literature. "
+        "Produce a concise literature review with 5-8 key citations in APA."
+    ),
+    tools=[format_citation],
+)
+
+methodology_designer = Agent(
+    name="Methodology Designer",
+    instructions=(
+        "You design rigorous research methodologies. "
+        "Output a clear, reproducible methods section."
+    ),
+    tools=[format_latex_section],
+)
+
+scientific_writer = build_scientific_writer()
+
+
+def run(topic: str) -> str:
+    """Run the 3-agent workflow for a given research ``topic``."""
+    t1 = Task(
+        name="review_literature",
+        description=f"Review literature on: {topic}",
+        agent=literature_reviewer,
+        expected_output="APA literature review.",
+    )
+    t2 = Task(
+        name="design_methodology",
+        description=f"Design research methodology for: {topic}",
+        agent=methodology_designer,
+        expected_output="LaTeX methods section.",
+    )
+    t3 = Task(
+        name="write_paper",
+        description=(
+            f"Combine the literature review and methodology into a full "
+            f"scientific paper on: {topic}. Use LaTeX formatting."
+        ),
+        agent=scientific_writer,
+        expected_output="Complete LaTeX paper.",
+    )
+    team = AgentTeam(
+        agents=[literature_reviewer, methodology_designer, scientific_writer],
+        tasks=[t1, t2, t3],
+    )
+    return team.start()
+
+
+if __name__ == "__main__":
+    result = run("Transformer architectures for protein folding prediction")
+    print(result)

examples/scientific_writing/scientific_writer.py

@@ -0,0 +1,85 @@
+"""Scientific Paper Writer — single-agent example using the CAJAL model.
+
+CAJAL (``Agnuxo/CAJAL-4B-P2PCLAW``) is a 4B-parameter HuggingFace model
+specialised for LaTeX-formatted academic writing.  This example shows how
+to use it with a regular ``Agent`` and ``@tool`` functions — no new Agent
+subclass is needed.  Run 3 ways: Python (this file), CLI, or YAML.
+
+Requirements:
+    pip install "praisonaiagents[llm]"          # for litellm / HuggingFace
+    export HUGGINGFACE_API_KEY=...              # or run a local inference server
+
+Usage:
+    python examples/scientific_writing/scientific_writer.py
+"""
+from praisonaiagents import Agent, tool
+
+
+@tool
+def format_latex_section(title: str, content: str) -> str:
+    """Wrap prose content in a LaTeX ``\\section{}`` block.
+
+    Args:
+        title: Section heading (e.g. ``"Introduction"``).
+        content: The body text.
+
+    Returns:
+        LaTeX-formatted section string.
+    """
+    return f"\\section{{{title}}}\n{content}\n"
+
+
+@tool
+def format_citation(authors: str, year: int, title: str, venue: str) -> str:
+    """Render a single APA-style citation line.
+
+    Args:
+        authors: ``"Smith, J. & Jones, A."``
+        year: Publication year.
+        title: Paper title.
+        venue: Journal / conference name.
+
+    Returns:
+        One APA-formatted citation string.
+    """
+    return f"{authors} ({year}). {title}. *{venue}*."
+
+
+SCIENTIFIC_INSTRUCTIONS = """You are a specialised scientific paper writer.
+
+Produce high-quality academic content with:
+- Clear structure (Abstract, Introduction, Methodology, Results, Discussion,
+  Conclusion, References)
+- Rigorous methodology and precise language
+- Proper APA citations and LaTeX formatting
+- Honest limitations and future work sections
+
+Use ``format_latex_section`` to wrap each section in LaTeX, and
+``format_citation`` to build the References list.
+"""
+
+
+def build_scientific_writer(
+    model: str = "huggingface/Agnuxo/CAJAL-4B-P2PCLAW",
+) -> Agent:
+    """Construct a scientific-writing Agent.
+
+    Override ``model`` for any HuggingFace / local / API model — the default
+    targets the CAJAL-4B checkpoint for LaTeX-style academic output.
+    """
+    return Agent(
+        name="Scientific Writer",
+        instructions=SCIENTIFIC_INSTRUCTIONS,
+        llm=model,
+        tools=[format_latex_section, format_citation],
+    )
+
+
+if __name__ == "__main__":
+    agent = build_scientific_writer()
+    paper = agent.start(
+        "Write a short scientific paper (Abstract + Introduction + Conclusion) "
+        "on 'Climate change effects on coral reef biodiversity'. "
+        "Use LaTeX formatting and include at least 3 APA citations."
+    )
+    print(paper)

examples/serve/mcp_http_server.py

@@ -44,12 +44,33 @@ def main():
     
     def calculate(expression: str) -> str:
         """Calculate a math expression safely."""
-        try:
-            # Safe eval for basic math
-            allowed = set("0123456789+-*/.() ")
-            if all(c in allowed for c in expression):
-                return str(eval(expression))
+        import ast
+
+        allowed = set("0123456789+-*/.() ")
+        if not all(c in allowed for c in expression):
             return "Error: Invalid expression"
+        try:
+            tree = ast.parse(expression, mode="eval")
+            for node in ast.walk(tree):
+                if not isinstance(
+                    node,
+                    (
+                        ast.Expression,
+                        ast.BinOp,
+                        ast.UnaryOp,
+                        ast.Constant,
+                        ast.Add,
+                        ast.Sub,
+                        ast.Mult,
+                        ast.Div,
+                        ast.USub,
+                        ast.UAdd,
+                    ),
+                ):
+                    return "Error: Invalid expression"
+            return str(
+                eval(compile(tree, "<expr>", "eval"), {"__builtins__": {}})
+            )
         except Exception as e:
             return f"Error: {e}"
     

examples/yaml/MULTI_CHANNEL_GUIDE.md

@@ -0,0 +1,222 @@
+# Multi-Channel Bot Deployment Guide
+
+This guide explains how to set up multiple bots on the same platform (e.g., multiple Telegram bots) for different roles using PraisonAI, similar to Hermes-style AI workforces.
+
+## Overview
+
+PraisonAI supports running multiple bots on the same platform, each routing to different agents with specialized knowledge and capabilities. This is useful for:
+
+- **Role-based bots**: `@company_cfo_bot`, `@company_ops_bot`, `@company_content_bot`
+- **Department-specific agents**: Each with different knowledge brains and tool access
+- **Workflow separation**: Different bots handle different types of requests
+
+## Quick Start with Onboard Wizard
+
+The easiest way to set up multiple channels is using the onboard wizard:
+
+```bash
+praisonai onboard
+```
+
+The wizard now supports:
+1. **Multi-channel setup**: Add multiple bots for the same platform
+2. **Role-based naming**: Automatically generates `TELEGRAM_<ROLE>_BOT_TOKEN` environment variables
+3. **Agent routing**: Each channel routes to a different agent
+
+### Onboard Wizard Flow
+
+1. **Choose platform**: Start with your first platform (e.g., telegram)
+2. **Add more channels**: Wizard asks "Add another bot channel?"
+3. **Specify role**: Enter role name (e.g., "cfo", "ops", "content")
+4. **Configure tokens**: Each channel gets its own environment variable
+5. **Test connections**: Wizard tests each bot separately
+
+## Environment Variable Convention
+
+For multiple bots on the same platform, use this naming convention:
+
+| Channel | Environment Variable | Bot Purpose |
+|---------|---------------------|-------------|
+| `telegram_cfo` | `TELEGRAM_CFO_BOT_TOKEN` | CFO-related tasks |
+| `telegram_ops` | `TELEGRAM_OPS_BOT_TOKEN` | Operations tasks |
+| `telegram_content` | `TELEGRAM_CONTENT_BOT_TOKEN` | Content creation |
+| `telegram_support` | `TELEGRAM_SUPPORT_BOT_TOKEN` | Customer support |
+
+### Setting Up Tokens
+
+1. **Create bots in @BotFather**:
+   ```text
+   /newbot
+   Bot name: Company CFO Bot
+   Username: @company_cfo_bot
+   -> Copy token
+   
+   /newbot
+   Bot name: Company Ops Bot
+   Username: @company_ops_bot
+   -> Copy token
+   ```
+
+2. **Set environment variables**:
+   ```bash
+   export TELEGRAM_CFO_BOT_TOKEN="123456789:ABC..."
+   export TELEGRAM_OPS_BOT_TOKEN="987654321:DEF..."
+   export TELEGRAM_CONTENT_BOT_TOKEN="456789123:GHI..."
+   ```
+
+3. **Or use `~/.praisonai/.env`**:
+   ```env
+   TELEGRAM_CFO_BOT_TOKEN=123456789:ABC...
+   TELEGRAM_OPS_BOT_TOKEN=987654321:DEF...
+   TELEGRAM_CONTENT_BOT_TOKEN=456789123:GHI...
+   TELEGRAM_ALLOWED_USERS=123456789,987654321
+   ```
+
+## Example Configuration
+
+See `multi-telegram-hermes-workforce.yaml` for a complete example.
+
+### Key Features
+
+1. **Multiple Channels**:
+   ```yaml
+   channels:
+     telegram_cfo:
+       platform: telegram
+       token: ${TELEGRAM_CFO_BOT_TOKEN}
+       routes:
+         default: cfo
+     
+     telegram_ops:
+       platform: telegram
+       token: ${TELEGRAM_OPS_BOT_TOKEN}
+       routes:
+         default: ops
+   ```
+
+2. **Specialized Agents**:
+   ```yaml
+   agents:
+     cfo:
+       instructions: "You are a CFO agent. Help with financial analysis..."
+       model: gpt-4o-mini
+     
+     ops:
+       instructions: "You are an ops agent. Help with system monitoring..."
+       model: gpt-4o-mini
+   ```
+
+## Validation with Doctor
+
+Use the doctor command to validate your multi-channel setup:
+
+```bash
+praisonai doctor
+```
+
+The doctor checks for:
+- ✅ **Unique tokens**: Each channel uses a different bot token
+- ✅ **Naming convention**: Follows `PLATFORM_ROLE_BOT_TOKEN` pattern  
+- ✅ **Missing tokens**: All referenced environment variables are set
+- ❌ **Duplicate tokens**: Same token used by multiple channels (fails)
+
+## Deployment
+
+### Start Gateway
+
+```bash
+# Start the multi-bot gateway
+praisonai gateway start --config multi-telegram-hermes-workforce.yaml
+
+# Or install as daemon
+praisonai onboard  # Wizard installs daemon automatically
+praisonai gateway status
+```
+
+### Verify Setup
+
+1. **Check health**: `curl http://127.0.0.1:8765/health`
+2. **View dashboard**: `praisonai claw` → http://127.0.0.1:8082
+3. **Test bots**: Send messages to each bot to verify routing
+
+## Common Patterns
+
+### Basic Multi-Role Setup
+```yaml
+# Single platform, multiple roles
+channels:
+  telegram_support:
+    token: ${TELEGRAM_SUPPORT_BOT_TOKEN}
+    routes:
+      default: support_agent
+  
+  telegram_sales:
+    token: ${TELEGRAM_SALES_BOT_TOKEN}
+    routes:
+      default: sales_agent
+```
+
+### Cross-Platform Workforce
+```yaml
+# Multiple platforms, multiple roles
+channels:
+  telegram_cfo:
+    platform: telegram
+    token: ${TELEGRAM_CFO_BOT_TOKEN}
+    routes:
+      default: cfo
+  
+  discord_ops:
+    platform: discord
+    token: ${DISCORD_OPS_BOT_TOKEN}
+    routes:
+      default: ops
+  
+  slack_content:
+    platform: slack
+    token: ${SLACK_CONTENT_BOT_TOKEN}
+    routes:
+      default: content
+```
+
+## Troubleshooting
+
+### Common Issues
+
+1. **"Token is used by multiple channels"**
+   - Each bot needs its own unique token from @BotFather
+   - Check environment variables for duplicates
+
+2. **"Channel skipped - no token"**
+   - Set the environment variable: `export TELEGRAM_CFO_BOT_TOKEN="..."`
+   - Or update `~/.praisonai/.env`
+
+3. **"Bot not responding"**
+   - Verify token is correct: `praisonai doctor`
+   - Check bot is not already running elsewhere
+   - Ensure allowlist includes your user ID
+
+### Debugging Commands
+
+```bash
+# Check configuration
+praisonai doctor
+
+# View logs
+praisonai gateway logs
+
+# Test connection
+curl -H "Authorization: Bearer $GATEWAY_AUTH_TOKEN" \
+     http://127.0.0.1:8765/info
+```
+
+## Migration from Single-Bot Setup
+
+If you currently have a single bot setup, you can add more channels:
+
+1. **Run onboard wizard**: `praisonai onboard`
+2. **Add new channels**: Wizard will preserve existing setup
+3. **Update configuration**: New channels added to existing `bot.yaml`
+4. **Restart gateway**: `praisonai gateway restart`
+
+The wizard preserves backward compatibility with single-bot configurations.

examples/yaml/a2ui-canvas.yaml

@@ -0,0 +1,19 @@
+# A2UI canvas agent — YAML example
+#
+# Usage:
+#   pip install praisonaiagents[a2ui]
+#   praisonai agent run --file a2ui-canvas.yaml --prompt "Show a welcome card"
+#
+# For live A2UI rendering, use PraisonAIUI (AG-UI / surfaces):
+#   pip install praisonaiui
+#   See examples/python/a2ui/send_surface.py and PraisonAIUI example 29.
+
+agent:
+  name: "ui-assistant"
+  instructions: |
+    You are a helpful assistant. When the user asks for UI elements
+    (buttons, cards, forms), call send_a2ui_messages with valid A2UI v0.9 JSON.
+    Use surface id "main" unless told otherwise.
+  llm: gpt-4o-mini
+  tools:
+    - send_a2ui_messages

examples/yaml/agent-with-mcp.yaml

@@ -0,0 +1,40 @@
+# Example: Agent with MCP Tool Integration
+#
+# This example shows how to configure an agent with MCP (Model Context Protocol)
+# servers for additional tool capabilities.
+#
+# Usage:
+#   praisonai workflow run --file examples/yaml/agent-with-mcp.yaml
+
+agent:
+  name: mcp_assistant
+  instructions: |
+    You are a helpful assistant with access to filesystem and time tools.
+    Use the MCP tools when needed to help users with file operations or time queries.
+  llm: gpt-4o-mini
+
+# MCP server configuration
+mcp:
+  servers:
+    filesystem:
+      # Filesystem access for the agent
+      command: npx
+      args:
+        - "-y"
+        - "@modelcontextprotocol/server-filesystem"
+        - "."
+      env: {}
+      enabled: true
+    
+    time:
+      # Time and date utilities
+      command: npx
+      args:
+        - "-y" 
+        - "@modelcontextprotocol/server-time"
+      env: {}
+      enabled: true
+
+# NOTE: Per-server tool filtering (tools.include/exclude) is not yet supported
+# by the MCP server config schema. Tool filtering should be applied after
+# calling mcp.get_tools() in your application code.

examples/yaml/message_steering_example.yaml

@@ -0,0 +1,47 @@
+framework: praisonai
+topic: AI Research with Message Steering
+
+roles:
+  researcher:
+    role: AI Research Analyst
+    goal: Conduct comprehensive AI research with real-time guidance
+    backstory: You are an expert AI researcher who can receive and incorporate real-time guidance during your work
+    message_steering: true  # Enable real-time message steering
+    tools:
+      - internet_search
+      - write_file
+    tasks:
+      research_task:
+        description: |
+          Research the latest developments in AI and write a comprehensive report.
+          You may receive additional guidance during execution via message steering.
+        expected_output: |
+          A well-structured research report covering recent AI developments,
+          incorporating any real-time guidance received during execution.
+
+  writer:
+    role: Technical Writer
+    goal: Write clear, engaging content based on research
+    backstory: You are a skilled technical writer who can adapt content based on real-time feedback
+    message_steering: true  # Enable steering for dynamic content adjustment
+    tools:
+      - read_file
+      - write_file
+    tasks:
+      writing_task:
+        description: |
+          Take the research findings and create an accessible article for a general audience.
+          Adjust style and focus based on any steering messages received.
+        expected_output: |
+          A polished article that transforms technical research into accessible content,
+          refined based on real-time steering guidance.
+
+dependencies:
+  - writing_task: [research_task]
+
+# Usage Instructions:
+# 1. Run: praisonai --agent-file message_steering_example.yaml
+# 2. While running, use steering API to send guidance:
+#    agent.steer("Focus more on practical applications")
+#    agent.steer("Keep it under 500 words", priority=15)
+#    agent.steer("Add more examples", priority=10)

examples/yaml/multi-telegram-hermes-workforce.yaml

@@ -0,0 +1,66 @@
+# Multi-Telegram Channel Hermes Workforce Example
+# Generated by `praisonai onboard` with multi-channel support
+# Start with: praisonai gateway start --config multi-telegram-hermes-workforce.yaml
+
+gateway:
+  host: "127.0.0.1"
+  port: 8765
+
+# Legacy single-agent block for backward compatibility
+agent:
+  name: "assistant"
+  instructions: "You are a helpful AI assistant. You can search the web, manage schedules, and remember past conversations. Use your tools proactively."
+  model: "gpt-4o-mini"
+
+# Multi-agent workforce with different specialized roles
+agents:
+  cfo:
+    instructions: "You are a CFO agent. You are a helpful AI assistant. Use your tools proactively to help with CFO-related tasks like financial analysis, budget planning, and cost optimization."
+    model: gpt-4o-mini
+    memory: false
+
+  ops:
+    instructions: "You are an ops agent. You are a helpful AI assistant. Use your tools proactively to help with ops-related tasks like system monitoring, deployment, and infrastructure management."
+    model: gpt-4o-mini
+    memory: false
+
+  content:
+    instructions: "You are a content agent. You are a helpful AI assistant. Use your tools proactively to help with content-related tasks like writing, editing, and content strategy."
+    model: gpt-4o-mini
+    memory: false
+
+# Multiple Telegram channels - each with unique tokens and routing
+channels:
+  telegram_cfo:
+    platform: telegram
+    token: ${TELEGRAM_CFO_BOT_TOKEN}
+    group_policy: mention_only
+    allowed_users: ${TELEGRAM_ALLOWED_USERS}
+    # auto_approve_tools: true
+                               # Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
+    routes:
+      default: cfo
+
+  telegram_ops:
+    platform: telegram
+    token: ${TELEGRAM_OPS_BOT_TOKEN}
+    group_policy: mention_only
+    allowed_users: ${TELEGRAM_ALLOWED_USERS}
+    # auto_approve_tools: true
+                               # Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
+    routes:
+      default: ops
+
+  telegram_content:
+    platform: telegram
+    token: ${TELEGRAM_CONTENT_BOT_TOKEN}
+    group_policy: mention_only
+    allowed_users: ${TELEGRAM_ALLOWED_USERS}
+    # auto_approve_tools: true
+                               # Chat bots cannot show CLI approval prompts; safe tools auto-approved by default.
+    routes:
+      default: content
+
+# Legacy routing block
+routing:
+  default: assistant

examples/yaml/toolsets_workflow.yaml

@@ -0,0 +1,148 @@
+# Named Toolsets YAML Example
+# This example demonstrates using toolsets in YAML configuration
+
+# Global configuration
+config:
+  llm: gpt-4o-mini
+  memory: true
+
+# Define agents with different toolsets
+agents:
+  # Research agent with research toolset (includes web tools)
+  researcher:
+    role: Senior Research Analyst
+    goal: Research topics thoroughly using web search and analysis tools
+    instructions: |
+      You are a senior research analyst with access to web search and file tools.
+      Research topics comprehensively and provide detailed findings.
+    toolsets:
+      - research  # Includes: web search, file tools, basic analysis
+    
+  # Writer agent with safe toolset for writing tasks
+  writer:
+    role: Content Writer
+    goal: Create engaging content based on research
+    instructions: |
+      You are a skilled content writer. Transform research into clear,
+      engaging content for the target audience.
+    toolsets:
+      - safe  # Safe, read-only tools for writing
+    
+  # Developer agent with development toolset
+  developer:
+    role: Software Developer
+    goal: Create and modify code based on requirements
+    instructions: |
+      You are a software developer with access to development tools.
+      Write, test, and debug code according to specifications.
+    toolsets:
+      - development  # Includes: code tools, file operations, testing
+      
+  # Data analyst with mixed tools and toolsets
+  analyst:
+    role: Data Analyst  
+    goal: Analyze data and generate insights
+    instructions: |
+      You are a data analyst. Process data files and generate
+      actionable insights and visualizations.
+    tools:
+      - read_file  # Explicit tool
+      - write_file  # Explicit tool
+    toolsets:
+      - safe  # Safe toolset for analysis
+      
+  # Security-focused agent with minimal toolset
+  security_auditor:
+    role: Security Auditor
+    goal: Review and audit for security issues
+    instructions: |
+      You are a security auditor with minimal tool access.
+      Focus on reading and analyzing without making changes.
+    toolsets:
+      - safe  # Minimal, safe tools only
+
+# Define tasks for the workflow
+tasks:
+  # Research task using research toolset
+  research_phase:
+    description: |
+      Research the latest trends in AI agent frameworks.
+      Focus on:
+      - Popular frameworks and their features
+      - Industry adoption patterns
+      - Security considerations
+      - Performance benchmarks
+    expected_output: |
+      A comprehensive research report covering AI agent frameworks,
+      including trends, adoption, security, and performance data.
+    agent: researcher
+    
+  # Content creation using safe toolset
+  content_creation:
+    description: |
+      Create a technical blog post based on the research findings.
+      The post should be:
+      - Well-structured with clear sections
+      - Accessible to technical and non-technical audiences  
+      - Include actionable insights
+      - 1000-1500 words
+    expected_output: |
+      A polished blog post about AI agent frameworks suitable
+      for publication on a technical blog.
+    agent: writer
+    
+  # Code development using development toolset
+  code_examples:
+    description: |
+      Create code examples demonstrating key concepts from the research.
+      Include:
+      - Simple "hello world" style examples
+      - Intermediate use cases
+      - Best practices examples
+      - Proper documentation
+    expected_output: |
+      A collection of well-documented code examples that illustrate
+      the concepts discussed in the research and blog post.
+    agent: developer
+    
+  # Data analysis using mixed tools/toolsets
+  trend_analysis:
+    description: |
+      Analyze the research data to identify trends and patterns.
+      Create:
+      - Summary statistics
+      - Trend analysis
+      - Recommendations based on findings
+    expected_output: |
+      A data analysis report with trends, patterns, and
+      actionable recommendations for AI framework adoption.
+    agent: analyst
+    
+  # Security review using minimal toolset
+  security_review:
+    description: |
+      Review all outputs for security considerations.
+      Check for:
+      - Potential security vulnerabilities in code examples
+      - Security best practices in recommendations
+      - Privacy considerations in data handling
+    expected_output: |
+      A security review report identifying any security concerns
+      and recommendations for secure implementation.
+    agent: security_auditor
+
+# Workflow process (sequential execution)
+process: sequential
+
+# Optional: Define workflow dependencies
+# tasks:
+#   research_phase:
+#     dependencies: []
+#   content_creation: 
+#     dependencies: [research_phase]
+#   code_examples:
+#     dependencies: [research_phase]
+#   trend_analysis:
+#     dependencies: [research_phase]
+#   security_review:
+#     dependencies: [content_creation, code_examples, trend_analysis]

src/praisonai-agents/.gitignore

@@ -95,4 +95,5 @@ chroma_db/
 .windsurf
 .ruff_cache
 *.sfdx
-.sfdx/
+.sfdx/
+.cursor/

src/praisonai-agents/AGENTS.md

@@ -378,7 +378,7 @@ Agent parameters are consolidated into Config objects following the pattern:
 
 | Config Object | Replaces (Deprecated) |
 |--------------|----------------------|
-| `execution=ExecutionConfig(code_execution=True, code_mode="safe", rate_limiter=obj)` | `allow_code_execution`, `code_execution_mode`, `rate_limiter` |
+| `execution=ExecutionConfig(code_execution=True, code_mode="safe", rate_limiter=obj, context_compaction=False)` | `allow_code_execution`, `code_execution_mode`, `rate_limiter` |
 | `memory=MemoryConfig(auto_save="session_name")` | `auto_save` |
 | `autonomy=AutonomyConfig(verification_hooks=[...])` | `verification_hooks` |
 | `handoffs=[other_agent]` | `allow_delegation` |
@@ -390,6 +390,8 @@ Agent parameters are consolidated into Config objects following the pattern:
 
 **Deprecated standalone params still work** (backward compatible) but emit `DeprecationWarning`.
 
+> **📢 Deprecation Notice:** `ExecutionConfig.context_compaction` currently defaults to `False` during a deprecation period. In the next release, it will default to `True` for automatic proactive context overflow protection. To opt out explicitly, set `context_compaction=False`. To use the new behavior early, set `context_compaction=True`.
+
 ```python
 # ❌ Old way (deprecated)
 agent = Agent(
@@ -413,6 +415,7 @@ agent = Agent(
         code_execution=True,
         code_mode="safe",
         rate_limiter=my_limiter,
+        context_compaction=False,  # prepare for next release when default becomes True; warning still fires during deprecation period
     ),
     memory=MemoryConfig(auto_save="my_session"),
     autonomy=AutonomyConfig(verification_hooks=[my_hook]),

src/praisonai-agents/benchmarks/real_benchmark.py

@@ -79,10 +79,20 @@ def get_time(timezone: Literal["est", "pst"]) -> str:
 
 
 def calculate(expression: str) -> str:
-    """Calculate a math expression."""
+    """Calculate a math expression safely."""
     try:
-        result = eval(expression)
-        return f"Result: {result}"
+        import ast, operator
+        _OPS = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
+                ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv,
+                ast.Mod: operator.mod, ast.Pow: operator.pow,
+                ast.USub: operator.neg, ast.UAdd: operator.pos}
+        def _ev(n):
+            if isinstance(n, ast.Expression): return _ev(n.body)
+            if isinstance(n, ast.Constant) and isinstance(n.value, (int, float)): return n.value
+            if isinstance(n, ast.UnaryOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.operand))
+            if isinstance(n, ast.BinOp) and type(n.op) in _OPS: return _OPS[type(n.op)](_ev(n.left), _ev(n.right))
+            raise ValueError(f"Unsupported: {ast.dump(n)}")
+        return f"Result: {_ev(ast.parse(expression, mode='eval'))}"
     except Exception:
         return "Error: Invalid expression"
 

src/praisonai-agents/examples/context_compaction_example.py

@@ -0,0 +1,229 @@
+#!/usr/bin/env python3
+"""
+Context Compaction Example - Python API
+
+Demonstrates how to configure context compaction policies in PraisonAI agents.
+This provides examples for CLI/YAML parity.
+"""
+
+from praisonaiagents import Agent
+from praisonaiagents.config.feature_configs import ExecutionConfig
+from praisonaiagents.context.adapters import ContextCompactionPolicyAdapter
+from praisonaiagents.context.protocols import CompactionStrategy
+
+
+def example_default_behavior():
+    """Example 1: Default behavior (during deprecation period)."""
+    print("=== Example 1: Default Behavior ===")
+    
+    # Currently defaults to False with deprecation warning
+    agent = Agent(
+        name="default_agent",
+        instructions="You are a helpful assistant."
+    )
+    
+    print(f"Context compaction enabled: {agent.execution.context_compaction}")
+    print("Note: Currently False during deprecation period, will change to True in next release\n")
+
+
+def example_explicit_disable():
+    """Example 2: Explicitly disable context compaction."""
+    print("=== Example 2: Explicitly Disabled ===")
+    
+    agent = Agent(
+        name="no_compaction_agent",
+        instructions="You are a helpful assistant.",
+        execution=ExecutionConfig(context_compaction=False)
+    )
+    
+    print(f"Context compaction enabled: {agent.execution.context_compaction}")
+    print("No automatic compaction will occur\n")
+
+
+def example_default_policy():
+    """Example 3: Use default compaction policy."""
+    print("=== Example 3: Default Policy ===")
+    
+    agent = Agent(
+        name="default_policy_agent", 
+        instructions="You are a helpful assistant.",
+        execution=ExecutionConfig(context_compaction=True)
+    )
+    
+    print(f"Context compaction enabled: {agent.execution.context_compaction}")
+    print("Will use balanced default policy (trigger at 90%, preserve 5 recent turns)\n")
+
+
+def example_conservative_policy():
+    """Example 4: Conservative compaction policy."""
+    print("=== Example 4: Conservative Policy ===")
+    
+    conservative_policy = ContextCompactionPolicyAdapter(
+        trigger_at=0.80,  # Trigger at 80% capacity
+        strategy=CompactionStrategy.DROP_OLDEST_TOOLS,
+        preserve_last_n_turns=8,  # Keep more conversation history
+        target_utilization=0.60,  # Target 60% utilization after compaction
+        aggressive_tool_truncation=False
+    )
+    
+    agent = Agent(
+        name="conservative_agent",
+        instructions="You are a helpful assistant.",
+        execution=ExecutionConfig(context_compaction=conservative_policy)
+    )
+    
+    print(f"Policy trigger: {conservative_policy.trigger_at}")
+    print(f"Strategy: {conservative_policy.strategy}")
+    print(f"Preserve turns: {conservative_policy.preserve_last_n_turns}")
+    print(f"Target utilization: {conservative_policy.target_utilization}\n")
+
+
+def example_aggressive_policy():
+    """Example 5: Aggressive compaction policy."""
+    print("=== Example 5: Aggressive Policy ===")
+    
+    aggressive_policy = ContextCompactionPolicyAdapter(
+        trigger_at=0.95,  # Wait until 95% capacity
+        strategy=CompactionStrategy.SUMMARISE,  # Use LLM summarization  
+        preserve_last_n_turns=3,  # Keep only recent exchanges
+        target_utilization=0.75,  # Target higher utilization 
+        aggressive_tool_truncation=True,  # Truncate large tool outputs
+        model_overrides={
+            "gpt-4": {"trigger_at": 0.92},  # Different threshold for gpt-4
+            "claude-3": {"trigger_at": 0.88}
+        }
+    )
+    
+    agent = Agent(
+        name="aggressive_agent",
+        instructions="You are a helpful assistant.",
+        execution=ExecutionConfig(context_compaction=aggressive_policy)
+    )
+    
+    print(f"Policy trigger: {aggressive_policy.trigger_at}")
+    print(f"Strategy: {aggressive_policy.strategy}")  
+    print(f"Preserve turns: {aggressive_policy.preserve_last_n_turns}")
+    print(f"Model overrides: {aggressive_policy.model_overrides}\n")
+
+
+def example_yaml_config():
+    """Example 6: YAML configuration equivalent."""
+    print("=== Example 6: YAML Configuration Equivalent ===")
+    print("""
+# agent_config.yaml
+name: "helpful_agent"
+instructions: "You are a helpful assistant."
+llm: "gpt-4o-mini"
+execution:
+  context_compaction:
+    trigger_at: 0.85
+    strategy: "drop_oldest_tools"
+    preserve_last_n_turns: 6
+    target_utilization: 0.65
+    aggressive_tool_truncation: true
+    model_overrides:
+      "gpt-4":
+        trigger_at: 0.90
+        strategy: "summarise"
+""")
+    
+    # Python equivalent:
+    policy = ContextCompactionPolicyAdapter(
+        trigger_at=0.85,
+        strategy="drop_oldest_tools",
+        preserve_last_n_turns=6,
+        target_utilization=0.65,
+        aggressive_tool_truncation=True,
+        model_overrides={
+            "gpt-4": {
+                "trigger_at": 0.90,
+                "strategy": "summarise"
+            }
+        }
+    )
+    
+    agent = Agent(
+        name="helpful_agent",
+        instructions="You are a helpful assistant.",
+        llm="gpt-4o-mini",
+        execution=ExecutionConfig(context_compaction=policy)
+    )
+    
+    print(f"Loaded policy trigger: {policy.trigger_at}")
+    print(f"Loaded strategy: {policy.strategy}\n")
+
+
+def example_cli_equivalent():
+    """Example 7: CLI equivalent commands."""
+    print("=== Example 7: CLI Equivalent ===")
+    print("""
+# Basic usage (will use default policy when enabled)
+praisonai agent run --name "assistant" --instructions "Be helpful"
+
+# Disable compaction explicitly  
+praisonai agent run --name "assistant" --execution '{"context_compaction": false}'
+
+# Use conservative policy
+praisonai agent run --name "assistant" \\
+    --execution '{
+        "context_compaction": {
+            "trigger_at": 0.80,
+            "strategy": "drop_oldest_tools",
+            "preserve_last_n_turns": 8
+        }
+    }'
+
+# Load from YAML config file
+praisonai agent run --config agent_config.yaml
+""")
+
+
+def example_serialization_round_trip():
+    """Example 8: Serialization round-trip."""
+    print("=== Example 8: Serialization Round-Trip ===")
+    
+    # Create agent with custom policy
+    original_policy = ContextCompactionPolicyAdapter(
+        trigger_at=0.88,
+        strategy=CompactionStrategy.SUMMARISE
+    )
+    
+    config = ExecutionConfig(context_compaction=original_policy)
+    
+    # Serialize to dict (for JSON/YAML storage)
+    config_dict = config.to_dict()
+    print(f"Serialized context_compaction type: {type(config_dict['context_compaction'])}")
+    
+    # Deserialize back to object
+    restored_config = ExecutionConfig.from_dict(config_dict)
+    restored_policy = restored_config.context_compaction
+    
+    print(f"Restored policy type: {type(restored_policy)}")
+    print(f"Attributes preserved: trigger_at={restored_policy.trigger_at}, strategy={restored_policy.strategy}")
+    print("✓ Round-trip serialization successful\n")
+
+
+if __name__ == "__main__":
+    print("Context Compaction Configuration Examples\n")
+    print("=" * 50)
+    
+    try:
+        example_default_behavior()
+        example_explicit_disable()
+        example_default_policy()
+        example_conservative_policy()
+        example_aggressive_policy()
+        example_yaml_config()
+        example_cli_equivalent()
+        example_serialization_round_trip()
+        
+        print("=" * 50)
+        print("✅ All examples completed successfully!")
+        print("\nFor more information, see:")
+        print("- Documentation: https://docs.praisonai.com/context-compaction")
+        print("- Examples: examples/context_compaction_example.py")
+        
+    except Exception as e:
+        print(f"❌ Example failed: {e}")
+        import traceback
+        traceback.print_exc()

src/praisonai-agents/praisonaiagents/__init__.py

@@ -127,6 +127,23 @@ _LAZY_IMPORTS = {
     'get_tool': ('praisonaiagents.tools.registry', 'get_tool'),
     'ToolRegistry': ('praisonaiagents.tools.registry', 'ToolRegistry'),
     
+    # Toolsets (named toolset groups)
+    'ToolsetSpec': ('praisonaiagents.toolsets', 'ToolsetSpec'),
+    'ToolsetRegistry': ('praisonaiagents.toolsets', 'ToolsetRegistry'),
+    'get_toolset_registry': ('praisonaiagents.toolsets', 'get_toolset_registry'),
+    'register_toolset': ('praisonaiagents.toolsets', 'register_toolset'),
+    'resolve_toolset': ('praisonaiagents.toolsets', 'resolve_toolset'),
+    'resolve_toolsets': ('praisonaiagents.toolsets', 'resolve_toolsets'),
+    'list_toolsets': ('praisonaiagents.toolsets', 'list_toolsets'),
+    'get_toolset': ('praisonaiagents.toolsets', 'get_toolset'),
+    'unregister_toolset': ('praisonaiagents.toolsets', 'unregister_toolset'),
+    'has_toolset': ('praisonaiagents.toolsets', 'has_toolset'),
+    
+    # Run outcomes - typed validation and agent execution results
+    'AgentRunOutcome': ('praisonaiagents.run_outcome', 'AgentRunOutcome'),
+    'RunStatus': ('praisonaiagents.run_outcome', 'RunStatus'),
+    'validate_decision_string': ('praisonaiagents.run_outcome', 'validate_decision_string'),
+    
     # Main display utilities (imports rich)
     'TaskOutput': ('praisonaiagents.main', 'TaskOutput'),
     'ReflectionOutput': ('praisonaiagents.main', 'ReflectionOutput'),
@@ -174,6 +191,7 @@ _LAZY_IMPORTS = {
     'RECOMMENDED_PROMPT_PREFIX': ('praisonaiagents.agent.handoff', 'RECOMMENDED_PROMPT_PREFIX'),
     'prompt_with_handoff_instructions': ('praisonaiagents.agent.handoff', 'prompt_with_handoff_instructions'),
     'HandoffConfig': ('praisonaiagents.agent.handoff', 'HandoffConfig'),
+    'HandoffToolPolicy': ('praisonaiagents.agent.handoff', 'HandoffToolPolicy'),
     'HandoffResult': ('praisonaiagents.agent.handoff', 'HandoffResult'),
     'HandoffInputData': ('praisonaiagents.agent.handoff', 'HandoffInputData'),
     'ContextPolicy': ('praisonaiagents.agent.handoff', 'ContextPolicy'),
@@ -343,6 +361,7 @@ _LAZY_IMPORTS = {
     # UI (optional)
     'AGUI': ('praisonaiagents.ui.agui', 'AGUI'),
     'A2A': ('praisonaiagents.ui.a2a', 'A2A'),
+    'A2UI': ('praisonaiagents.ui.a2ui', 'A2UI'),
     
     # Feature configs
     'MemoryConfig': ('praisonaiagents.config.feature_configs', 'MemoryConfig'),
@@ -383,6 +402,18 @@ _LAZY_IMPORTS = {
     'MultiAgentExecutionConfig': ('praisonaiagents.config.feature_configs', 'MultiAgentExecutionConfig'),
     'MultiAgentPlanningConfig': ('praisonaiagents.config.feature_configs', 'MultiAgentPlanningConfig'),
     'MultiAgentMemoryConfig': ('praisonaiagents.config.feature_configs', 'MultiAgentMemoryConfig'),
+    'ToolSearchConfig': ('praisonaiagents.config.feature_configs', 'ToolSearchConfig'),
+    
+    # Context compaction policies
+    'ContextCompactionPolicy': ('praisonaiagents.context.policy', 'ContextCompactionPolicy'),
+    'ContextCompactionPolicyProtocol': ('praisonaiagents.context.policy', 'ContextCompactionPolicyProtocol'),
+    'CompactionRoute': ('praisonaiagents.context.policy', 'CompactionRoute'),
+    'CompactionStrategy': ('praisonaiagents.context.policy', 'CompactionStrategy'),
+    'ContextBudgetResult': ('praisonaiagents.context.policy', 'ContextBudgetResult'),
+    'get_default_policy': ('praisonaiagents.context.policy', 'get_default_policy'),
+    'CONSERVATIVE_POLICY': ('praisonaiagents.context.policy', 'CONSERVATIVE_POLICY'),
+    'BALANCED_POLICY': ('praisonaiagents.context.policy', 'BALANCED_POLICY'),
+    'AGGRESSIVE_POLICY': ('praisonaiagents.context.policy', 'AGGRESSIVE_POLICY'),
     
     # Parameter resolver
     'resolve': ('praisonaiagents.config.param_resolver', 'resolve'),
@@ -624,6 +655,11 @@ def _custom_handler(name, cache):
         mod = importlib.import_module('.db', 'praisonaiagents')
         cache['db'] = mod
         return mod
+    if name == 'toolsets':
+        import importlib
+        mod = importlib.import_module('.toolsets', 'praisonaiagents')
+        cache['toolsets'] = mod
+        return mod
     
     raise AttributeError(f"Not handled by custom_handler: {name}")
 
@@ -814,6 +850,6 @@ def __dir__():
 # - Workflow, Task, Route, Parallel, Loop, Repeat, Pipeline, etc.
 # - MCP, FlowDisplay, track_workflow, FastContext, etc.
 # - Plan, PlanStep, TodoList, PlanningAgent, ApprovalCallback, etc.
-# - RAG, RAGConfig, RAGResult, AGUI, A2A, etc.
+# - RAG, RAGConfig, RAGResult, AGUI, A2A, A2UI, etc.
 # - All telemetry, display, and utility functions
 # ============================================================================

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

@@ -114,11 +114,12 @@ def __getattr__(name):
     
     # Handoff - lightweight
     _handoff_names = {
-        'Handoff', 'handoff', 'handoff_filters', 'parallel_handoffs',
+        'Handoff', 'handoff_filters', 'parallel_handoffs',
         'RECOMMENDED_PROMPT_PREFIX', 'prompt_with_handoff_instructions',
         'HandoffConfig', 'HandoffResult', 'HandoffInputData',
         'ContextPolicy', 'HandoffError', 'HandoffCycleError', 
-        'HandoffDepthError', 'HandoffTimeoutError'
+        'HandoffDepthError', 'HandoffTimeoutError', 'HandoffValidationError',
+        'TypedHandoff'
     }
     if name in _handoff_names:
         from . import handoff as _handoff_module
@@ -165,6 +166,21 @@ def __getattr__(name):
         _lazy_cache[name] = value
         return value
     
+    # Message Steering - lightweight protocols and implementations
+    _steering_names = {
+        'MessageSteeringProtocol', 'AgentSteeringProtocol', 'SteeringPriority',
+        'SteeringMessage', 'MessageSteering', 'SteeringMixin'
+    }
+    if name in _steering_names:
+        if name in {'MessageSteeringProtocol', 'AgentSteeringProtocol', 'SteeringPriority', 'SteeringMessage'}:
+            from . import protocols as _protocols_module
+            value = getattr(_protocols_module, name)
+        else:
+            from . import message_steering as _steering_module
+            value = getattr(_steering_module, name)
+        _lazy_cache[name] = value
+        return value
+    
     # Decomposed mixins - for advanced use cases
     if name == 'ToolExecutionMixin':
         from .tool_execution import ToolExecutionMixin
@@ -230,6 +246,8 @@ __all__ = [
     'HandoffCycleError',
     'HandoffDepthError',
     'HandoffTimeoutError',
+    'HandoffValidationError',
+    'TypedHandoff',
     'RouterAgent',
     'DeepResearchAgent',
     'DeepResearchResponse',
@@ -253,6 +271,13 @@ __all__ = [
     'MemoryAwareAgentProtocol',
     'FullAgentProtocol',
     'ContextEngineerProtocol',
+    # Message Steering
+    'MessageSteeringProtocol',
+    'AgentSteeringProtocol',
+    'SteeringPriority',
+    'SteeringMessage',
+    'MessageSteering',
+    'SteeringMixin',
     # Decomposed mixins (for advanced use cases)
     'ToolExecutionMixin',
     'ChatHandlerMixin',

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

@@ -21,6 +21,8 @@ from .chat_handler import ChatHandlerMixin
 from .session_manager import SessionManagerMixin
 from .async_safety import AsyncSafeState
 from .unified_execution_mixin import UnifiedExecutionMixin
+from .sandbox_mixin import SandboxMixin
+from .message_steering import SteeringMixin
 
 # Module-level logger for thread safety errors and debugging
 logger = get_logger(__name__)
@@ -241,7 +243,7 @@ def _get_default_server_registry() -> ServerRegistry:
 
 if TYPE_CHECKING:
     from ..approval.protocols import ApprovalConfig, ApprovalProtocol
-    from ..config.feature_configs import LearnConfig, MemoryConfig
+    from ..config.feature_configs import LearnConfig, MemoryConfig, ToolConfig
     from ..context.models import ContextConfig
     from ..context.manager import ContextManager
     from ..knowledge.knowledge import Knowledge
@@ -251,11 +253,12 @@ if TYPE_CHECKING:
     from .handoff import Handoff, HandoffConfig, HandoffResult
     from ..rag.models import RAGResult, ContextPack
     from ..eval.results import EvaluationLoopResult
+    from ..tools.retry import RetryPolicy
 
 # Import structured error from central errors module
 from ..errors import BudgetExceededError
 
-class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, SessionManagerMixin, ChatMixin, ExecutionMixin, MemoryMixin, AsyncMemoryMixin):
+class Agent(SteeringMixin, SandboxMixin, UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, SessionManagerMixin, ChatMixin, ExecutionMixin, MemoryMixin, AsyncMemoryMixin):
     # Class-level counter for generating unique display names for nameless agents
     _agent_counter = 0
     _agent_counter_lock = threading.Lock()
@@ -522,6 +525,7 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
         auth: Optional[str] = None,  # Subscription auth provider: "claude-code", "codex", etc.
         # Tools
         tools: Optional[List[Any]] = None,
+        toolsets: Optional[List[str]] = None,  # Named toolset groups to resolve
         allow_delegation: bool = False,  # Deprecated: use handoffs= instead
         allow_code_execution: Optional[bool] = False,  # Deprecated: use execution=ExecutionConfig(code_execution=True)
         code_execution_mode: Literal["safe", "unsafe"] = "safe",  # Deprecated: use execution=ExecutionConfig(code_mode="safe")
@@ -549,12 +553,17 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
         hooks: Optional[Union[List[Any], Dict[str, Any], 'HooksConfig']] = None,
         skills: Optional[Union[List[str], str, Dict[str, Any], 'SkillsConfig']] = None,
         approval: Optional[Union[bool, str, Dict[str, Any], 'ApprovalConfig', 'ApprovalProtocol']] = None,
-        tool_timeout: Optional[int] = None,  # P8/G11: Timeout in seconds for each tool call
-        parallel_tool_calls: bool = False,  # Gap 2: Enable parallel execution of batched LLM tool calls
+        tool_config: Optional[Union[bool, 'ToolConfig']] = None,  # Tool execution configuration (timeout, retry, parallel)
+        tool_timeout: Optional[int] = None,  # **Deprecated**: use tool_config=ToolConfig(timeout=60)
+        tool_retry_policy: Optional['RetryPolicy'] = None,  # **Deprecated**: use tool_config=ToolConfig(retry_policy=policy)
+        parallel_tool_calls: bool = False,  # **Deprecated**: use tool_config=ToolConfig(parallel=True)
         learn: Optional[Union[bool, str, Dict[str, Any], 'LearnConfig']] = None,  # Continuous learning (peer to memory)
         backend: Optional[Any] = None,  # External managed agent backend (e.g., ManagedAgentIntegration)
         cli_backend: Optional[Union[str, Any]] = None,  # CLI backend for delegating turns (e.g., "claude-code")
         interrupt_controller: Optional['InterruptController'] = None,  # G2: Cooperative cancellation
+        tool_search: Optional[Union[bool, str, Dict[str, Any], 'ToolSearchConfig']] = False,  # Progressive tool disclosure
+        message_steering: Optional[Union[bool, 'MessageSteeringProtocol']] = False,  # Real-time message steering during execution
+        sandbox: Optional[Union[bool, 'SandboxConfig']] = None,  # Sandbox for safe code execution
     ):
         """Initialize an Agent instance.
 
@@ -642,10 +651,14 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
                 - LearnConfig: Custom configuration
                 Learning is a first-class citizen, peer to memory. It captures patterns,
                 preferences, and insights from interactions to improve future responses.
-            parallel_tool_calls: Enable parallel execution of batched LLM tool calls (default False).
-                When True and LLM returns multiple tool calls in a single response, they execute
-                concurrently instead of sequentially. Provides ~3x speedup for I/O-bound tools.
-                Maintains backward compatibility with False default.
+            tool_config: Tool execution configuration. Accepts:
+                - bool: True enables defaults, False disables
+                - ToolConfig: Custom configuration for timeout, retry policy, parallel execution
+                Consolidates tool_timeout, tool_retry_policy, and parallel_tool_calls settings.
+            tool_retry_policy: **Deprecated** — use ``tool_config=ToolConfig(retry_policy=policy)`` instead.
+                Still works for backward compatibility.
+            parallel_tool_calls: **Deprecated** — use ``tool_config=ToolConfig(parallel=True)`` instead.
+                Still works for backward compatibility.
             backend: External managed agent backend for hybrid execution. Accepts:
                 - ManagedAgentIntegration: External managed agent service
                 - None: Use local execution (default)
@@ -658,6 +671,14 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
                 When provided, agent delegates entire conversation turns to the CLI tool
                 instead of using the built-in LLM. Enables session continuity and 
                 tool integration through external AI coding assistants.
+            tool_search: Progressive tool disclosure configuration. Accepts:
+                - bool: False=disabled (default), True=auto mode
+                - str: Mode ("auto", "on", "off")  
+                - Dict[str, Any]: Config overrides (e.g. {"threshold_pct": 15})
+                - ToolSearchConfig: Custom configuration
+                When enabled, replaces large tool schemas with bridge tools (tool_search,
+                tool_describe, tool_call) to save context. Core SDK tools never defer.
+                Auto mode activates based on token threshold. Opt-in feature.
 
         Raises:
             ValueError: If all of name, role, goal, backstory, and instructions are None.
@@ -677,6 +698,7 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
             - auto_save → memory=MemoryConfig(auto_save=)
             - rate_limiter → execution=ExecutionConfig(rate_limiter=)
             - verification_hooks → autonomy=AutonomyConfig(verification_hooks=)
+            - tool_timeout, tool_retry_policy, parallel_tool_calls → tool_config=ToolConfig(...)
         """
         # Add check at start if memory is requested
         if memory is not None:
@@ -1428,12 +1450,45 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
             web_search = False
             web_fetch = False
         
+        # ─────────────────────────────────────────────────────────────────────
+        # Resolve TOOL_SEARCH param
+        # ─────────────────────────────────────────────────────────────────────
+        # Fast path: None/False -> disabled (zero overhead)
+        if tool_search is None or tool_search is False:
+            self._tool_search_config = None
+        elif tool_search is True:
+            # True -> auto mode with defaults
+            from ..tools.tool_search import ToolSearchConfig as _ToolSearchConfig
+            self._tool_search_config = _ToolSearchConfig(enabled="auto")
+        elif isinstance(tool_search, str):
+            # String mode ("auto", "on", "off")
+            from ..tools.tool_search import ToolSearchConfig as _ToolSearchConfig
+            self._tool_search_config = _ToolSearchConfig.from_raw(tool_search)
+        elif isinstance(tool_search, dict):
+            # Dict -> config overrides
+            from ..tools.tool_search import ToolSearchConfig as _ToolSearchConfig
+            self._tool_search_config = _ToolSearchConfig(**tool_search)
+        else:
+            from ..tools.tool_search import ToolSearchConfig as _ToolSearchConfig
+            if isinstance(tool_search, _ToolSearchConfig):
+                self._tool_search_config = tool_search
+            else:
+                raise TypeError(
+                    "tool_search must be False/None, True, a mode string, "
+                    "a dict of ToolSearchConfig fields, or ToolSearchConfig"
+                )
+        
         # ============================================================
         # END CONSOLIDATED PARAMS EXTRACTION
         # ============================================================
         
         # Initialize autonomy features (agent-centric escalation/doom-loop)
         self._init_autonomy(autonomy, verification_hooks=verification_hooks)
+        
+        # Initialize loop guard for all execution modes (not just autonomous)
+        # Provides idempotency-aware guardrails for tool execution loops
+        from ..escalation.loop_guard import LoopGuard, LoopGuardConfig
+        self._loop_guard = LoopGuard(LoopGuardConfig(enabled=True))
 
         # If instructions are provided, use them to set role, goal, and backstory
         if instructions:
@@ -1467,8 +1522,17 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
             self.self_reflect = True if self_reflect is None else self_reflect
         
         self.instructions = instructions
+        
+        # Resolve tool_config early for immediate use in parallel_tool_calls
+        _tool_config = Agent._resolve_tool_config(
+            tool_config=tool_config,
+            tool_timeout=tool_timeout,
+            tool_retry_policy=tool_retry_policy, 
+            parallel_tool_calls=parallel_tool_calls
+        )
+        
         # Gap 2: Store parallel tool calls setting for ToolCallExecutor selection
-        self.parallel_tool_calls = parallel_tool_calls
+        self.parallel_tool_calls = _tool_config.parallel if _tool_config else parallel_tool_calls
         # G2: Store interrupt controller for cooperative cancellation
         self.interrupt_controller = interrupt_controller
         # Check for model name in environment variable if not provided
@@ -1525,6 +1589,7 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
                     if auth:
                         llm_config['auth'] = auth
                     llm_config['metrics'] = metrics
+                    llm_config['max_iter'] = max_iter
                     self.llm_instance = LLM(**llm_config)
                     self.llm = llm.get('model', Agent._get_default_model())
                 else:
@@ -1536,6 +1601,7 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
                         api_key=api_key,
                         auth=auth,
                         metrics=metrics,
+                        max_iter=max_iter,
                         web_search=web_search,
                         web_fetch=web_fetch,
                         prompt_caching=prompt_caching,
@@ -1563,6 +1629,7 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
                 # Add metrics parameter
                 llm = llm.copy()
                 llm['metrics'] = metrics
+                llm['max_iter'] = max_iter
                 self.llm_instance = LLM(**llm)  # Pass all dict items as kwargs
                 self._using_custom_llm = True
                 self.llm = llm.get('model', Agent._get_default_model())
@@ -1586,6 +1653,7 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
                 llm_params['web_fetch'] = web_fetch
                 llm_params['prompt_caching'] = prompt_caching
                 llm_params['claude_memory'] = claude_memory
+                llm_params['max_iter'] = max_iter
                 self.llm_instance = LLM(**llm_params)
                 self._using_custom_llm = True
                 self.llm = llm
@@ -1619,6 +1687,32 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
         else:
             self.tools = tools or []
         
+        # Handle toolsets parameter - resolve named toolset groups
+        if toolsets:
+            try:
+                from ..toolsets import resolve_toolsets
+                toolset_tool_names = resolve_toolsets(toolsets)
+                # Remove duplicates with existing tools
+                existing_tool_names = set()
+                for tool in self.tools:
+                    name = getattr(tool, 'name', getattr(tool, '__name__', str(tool)))
+                    existing_tool_names.add(name)
+                
+                unique_tool_names = [name for name in toolset_tool_names if name not in existing_tool_names]
+                toolset_tools = self._resolve_tool_names(unique_tool_names)
+                self.tools.extend(toolset_tools)
+                logging.debug(f"Resolved toolsets {toolsets} to {len(toolset_tools)} tools: {[getattr(t, '__name__', str(t)) for t in toolset_tools]}")
+            except (ValueError, KeyError) as e:
+                raise ValueError(
+                    f"Agent '{getattr(self, 'display_name', 'unknown')}' failed to resolve toolsets {toolsets}: {e}. "
+                    "Verify names with praisonaiagents.toolsets.list_toolsets()."
+                ) from e
+            except ImportError as e:
+                raise ImportError(
+                    f"Agent '{getattr(self, 'display_name', 'unknown')}' failed to import toolsets module: {e}. "
+                    "Ensure praisonaiagents is properly installed."
+                ) from e
+        
         # Inject default tools for autonomy mode (after self.tools is initialized)
         # ONLY inject if caller didn't provide tools - avoid duplicates with CLI/wrapper tools
         if self.autonomy_enabled and hasattr(self, '_autonomy_config_obj'):
@@ -1654,6 +1748,7 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
         self.max_execution_time = max_execution_time
         self._memory_instance = None
         self._init_memory(memory, user_id)
+        self._init_message_steering(message_steering)
         self.verbose = verbose
         self._has_explicit_output_config = _has_explicit_output  # Track if user set output mode
         self.tool_output_limit = tool_output_limit  # Configurable tool output limit
@@ -1678,6 +1773,9 @@ class Agent(UnifiedExecutionMixin, ToolExecutionMixin, ChatHandlerMixin, Session
         self.embedder_config = embedder_config
         self.knowledge = knowledge
         self.use_system_prompt = use_system_prompt
+        
+        # Store ExecutionConfig for context compaction policy access
+        self.execution = _exec_config
         # Async-safe chat_history with dual-lock protection
         self.__chat_history_state = AsyncSafeState([])
         
@@ -1833,8 +1931,14 @@ Your Goal: {self.goal}
         self._pending_approvals = {}
         self._approvals_lock = asyncio.Lock()
         
+        # Store the resolved tool_config for cloning (resolved earlier in __init__)
+        self._tool_config = _tool_config
+        
         # P8/G11: Tool timeout - prevent slow tools from blocking
-        self._tool_timeout = tool_timeout
+        self._tool_timeout = _tool_config.timeout if _tool_config else tool_timeout
+        
+        # Store tool retry policy for tool execution with exponential backoff
+        self._tool_retry_policy = _tool_config.retry_policy if _tool_config else tool_retry_policy
         
         # Cache for system prompts and formatted tools with eager thread-safe lock
         # Use OrderedDict for LRU behavior
@@ -1946,6 +2050,172 @@ Your Goal: {self.goal}
         # Telemetry - lazy initialized via property for performance
         self.__telemetry = None
         self.__telemetry_initialized = False
+        
+        # Sandbox configuration - initialize SandboxMixin
+        super().__init__(sandbox=sandbox)
+
+    @staticmethod
+    def _resolve_tool_config(tool_config, tool_timeout, tool_retry_policy, parallel_tool_calls):
+        """Resolve tool_config parameter with backward compatibility."""
+        # Import here to avoid circular imports
+        from ..config.feature_configs import ToolConfig, resolve_tools
+        
+        # Handle the new consolidated parameter
+        if tool_config is not None:
+            resolved_config = resolve_tools(tool_config)
+            if resolved_config:
+                # Apply deprecated parameter overrides if specified
+                if tool_timeout is not None:
+                    resolved_config.timeout = tool_timeout
+                if tool_retry_policy is not None:  
+                    resolved_config.retry_policy = tool_retry_policy
+                if parallel_tool_calls:
+                    resolved_config.parallel = parallel_tool_calls
+                return resolved_config
+        
+        # Fallback: create config from deprecated parameters if any are specified
+        if tool_timeout is not None or tool_retry_policy is not None or parallel_tool_calls:
+            import warnings
+            warnings.warn(
+                "tool_timeout, tool_retry_policy, and parallel_tool_calls are deprecated. "
+                "Use tool_config=ToolConfig(timeout=60, retry_policy=policy, parallel=True) instead.",
+                DeprecationWarning,
+                stacklevel=4  # Increased to account for static method call
+            )
+            return ToolConfig(
+                timeout=tool_timeout,
+                retry_policy=tool_retry_policy,
+                parallel=parallel_tool_calls
+            )
+        
+        return None
+
+    def __deepcopy__(self, memo: dict) -> "Agent":
+        """Custom deepcopy that creates fresh threading primitives.
+
+        threading.RLock (self.__cache_lock) and threading.Lock (self._cost_lock)
+        cannot be deep-copied on CPython < 3.13.  This hook deep-copies every
+        other attribute normally and replaces the locks with new instances so
+        that copy.deepcopy(agent) works in any Python version.
+        """
+        import copy
+        cls = self.__class__
+        result = cls.__new__(cls)
+        memo[id(self)] = result
+        for k, v in self.__dict__.items():
+            if k in ("_Agent__cache_lock",):
+                object.__setattr__(result, k, threading.RLock())
+            elif k == "_cost_lock":
+                object.__setattr__(result, k, threading.Lock())
+            else:
+                object.__setattr__(result, k, copy.deepcopy(v, memo))
+        return result
+
+    def clone_for_channel(self) -> "Agent":
+        """Return a fully independent copy of this agent for a gateway channel.
+        
+        This method safely clones an agent for use in multi-channel scenarios
+        by avoiding deepcopy issues with tools/handoffs and creating fresh
+        instances of mutable objects like interrupt_controller.
+        
+        Returns:
+            Agent: A new Agent instance with copied configuration but fresh
+                  locks, interrupt controller, and shallow-copied tools.
+        """
+        import copy
+        from ..tools.base import BaseTool
+        
+        # Build kwargs for new Agent instance, copying core attributes
+        clone_kwargs = {
+            # Core identity
+            'name': self.name,
+            'role': self.role, 
+            'goal': self.goal,
+            'backstory': self.backstory,
+            'instructions': self.instructions,
+            
+            # LLM configuration 
+            'llm': self.llm,
+            'base_url': getattr(self, 'base_url', None),
+            'api_key': getattr(self, 'api_key', None),
+            'auth': getattr(self, 'auth', None),
+            
+            # Shallow copy tools to avoid deepcopy issues with nested objects
+            'tools': list(self.tools) if self.tools else None,
+            
+            # Skip handoffs entirely - they shouldn't be shared across channels
+            # and can contain nested Agent instances that cause RLock issues
+            'handoffs': None,
+            
+            # Feature configurations - check for actual stored config objects
+            'memory': getattr(self, '_memory_config', None),
+            'knowledge': getattr(self, '_knowledge_config', None), 
+            'planning': getattr(self, '_planning_config', None),
+            'reflection': getattr(self, '_reflection_config', None),
+            'guardrails': getattr(self, '_guardrails_config', None),
+            'web': getattr(self, '_web_config', None),
+            'context': getattr(self, '_context_config', None),
+            'autonomy': getattr(self, '_autonomy_config', None),
+            'output': getattr(self, '_output_config', None),
+            'execution': getattr(self, '_execution_config', None),
+            'templates': getattr(self, '_template_config', None),
+            'caching': getattr(self, '_caching_config', None),
+            'hooks': getattr(self, '_hooks_config', None),
+            'skills': getattr(self, '_skills_config', None),
+            'approval': getattr(self, '_approval_config', None),
+            'learn': getattr(self, '_learn_config', None),
+            'tool_search': getattr(self, '_tool_search_config', None),
+            
+            # Tool configuration - use consolidated config when available  
+            'tool_config': getattr(self, '_tool_config', None),
+            # Backward compatibility - always include for now
+            'tool_timeout': getattr(self, '_tool_timeout', None),
+            'tool_retry_policy': getattr(self, '_tool_retry_policy', None),
+            'parallel_tool_calls': getattr(self, 'parallel_tool_calls', False),
+            
+            # CLI backend
+            'cli_backend': getattr(self, '_cli_backend', None),
+            
+            # Create fresh interrupt controller to avoid shared state
+            'interrupt_controller': None,  # Let new instance create its own
+            
+            # Sandbox config
+            'sandbox': getattr(self, '_sandbox_config', None),
+        }
+        
+        # Handle deprecated parameters for backward compatibility
+        import warnings
+        
+        # Check for deprecated standalone attributes and emit warnings
+        deprecated_attrs = [
+            ('allow_code_execution', 'execution'),
+            ('code_execution_mode', 'execution'), 
+            ('auto_save', 'memory'),
+            ('rate_limiter', 'execution'),
+            ('allow_delegation', 'handoffs'),
+            ('verification_hooks', 'autonomy')
+        ]
+        
+        for old_attr, new_param in deprecated_attrs:
+            if hasattr(self, old_attr):
+                value = getattr(self, old_attr)
+                if value is not None and value != False:  # Skip None/False defaults
+                    warnings.warn(
+                        f"Deprecated attribute '{old_attr}' found in agent clone. "
+                        f"Use '{new_param}=' parameter instead.",
+                        DeprecationWarning,
+                        stacklevel=2
+                    )
+                    # For basic compatibility, include in clone_kwargs if not already set
+                    if old_attr == 'allow_code_execution' and clone_kwargs['execution'] is None:
+                        from ..config.execution import ExecutionConfig
+                        clone_kwargs['execution'] = ExecutionConfig(code_execution=value)
+                    elif old_attr == 'auto_save' and clone_kwargs['memory'] is None:
+                        from ..config.memory import MemoryConfig  
+                        clone_kwargs['memory'] = MemoryConfig(auto_save=value)
+        
+        # Create new Agent instance
+        return self.__class__(**{k: v for k, v in clone_kwargs.items() if v is not None})
 
     @property
     def _telemetry(self):
@@ -4079,33 +4349,95 @@ Summary:"""
         
         return ""
     
-    def store_memory(self, content: str, memory_type: str = "short_term", **kwargs: Any) -> None:
+    def store_memory(self, content: str, memory_type: str = "short_term", action: str = "add", **kwargs: Any) -> None:
         """
         Store content in memory.
         
         Args:
             content: Content to store
             memory_type: Type of memory (short_term, long_term, entity, episodic)
+            action: Type of operation ("add", "replace", "remove")
             **kwargs: Additional arguments for the memory method
         """
         if not self._memory_instance:
             return
         
-        # Use protocol names first (store_*), fallback to legacy names (add_*)
-        if memory_type == "short_term":
-            if hasattr(self._memory_instance, 'store_short_term'):
-                self._memory_instance.store_short_term(content, **kwargs)
-            elif hasattr(self._memory_instance, 'add_short_term'):
-                self._memory_instance.add_short_term(content, **kwargs)
-        elif memory_type == "long_term":
-            if hasattr(self._memory_instance, 'store_long_term'):
-                self._memory_instance.store_long_term(content, **kwargs)
-            elif hasattr(self._memory_instance, 'add_long_term'):
-                self._memory_instance.add_long_term(content, **kwargs)
-        elif memory_type == "entity" and hasattr(self._memory_instance, 'add_entity'):
-            self._memory_instance.add_entity(content, **kwargs)
-        elif memory_type == "episodic" and hasattr(self._memory_instance, 'add_episodic'):
-            self._memory_instance.add_episodic(content, **kwargs)
+        # Validate memory_type
+        supported_types = ["short_term", "long_term", "entity", "episodic"]
+        if memory_type not in supported_types:
+            raise ValueError(f"Unsupported memory_type '{memory_type}'. Supported: {supported_types}")
+        
+        # Perform the actual storage operation first
+        storage_success = False
+        try:
+            if action == "add":
+                # Use protocol names first (store_*), fallback to legacy names (add_*)
+                if memory_type == "short_term":
+                    if hasattr(self._memory_instance, 'store_short_term'):
+                        self._memory_instance.store_short_term(content, **kwargs)
+                        storage_success = True
+                    elif hasattr(self._memory_instance, 'add_short_term'):
+                        self._memory_instance.add_short_term(content, **kwargs)
+                        storage_success = True
+                elif memory_type == "long_term":
+                    if hasattr(self._memory_instance, 'store_long_term'):
+                        self._memory_instance.store_long_term(content, **kwargs)
+                        storage_success = True
+                    elif hasattr(self._memory_instance, 'add_long_term'):
+                        self._memory_instance.add_long_term(content, **kwargs)
+                        storage_success = True
+                elif memory_type == "entity" and hasattr(self._memory_instance, 'add_entity'):
+                    self._memory_instance.add_entity(content, **kwargs)
+                    storage_success = True
+                elif memory_type == "episodic" and hasattr(self._memory_instance, 'add_episodic'):
+                    self._memory_instance.add_episodic(content, **kwargs)
+                    storage_success = True
+            elif action == "replace":
+                # For replace operations - look for replace_ or update_ methods
+                method_name = f"replace_{memory_type}" if hasattr(self._memory_instance, f'replace_{memory_type}') else f"update_{memory_type}"
+                if hasattr(self._memory_instance, method_name):
+                    getattr(self._memory_instance, method_name)(content, **kwargs)
+                    storage_success = True
+                else:
+                    raise ValueError(f"Memory provider does not support 'replace' for {memory_type}")
+            elif action == "remove":
+                # For remove operations - look for remove_ or delete_ methods  
+                method_name = f"remove_{memory_type}" if hasattr(self._memory_instance, f'remove_{memory_type}') else f"delete_{memory_type}"
+                if hasattr(self._memory_instance, method_name):
+                    getattr(self._memory_instance, method_name)(content, **kwargs)
+                    storage_success = True
+                else:
+                    raise ValueError(f"Memory provider does not support 'remove' for {memory_type}")
+            else:
+                raise ValueError(f"Unsupported action '{action}'. Supported: add, replace, remove")
+                
+            if not storage_success:
+                raise ValueError(f"Memory provider does not support {memory_type} storage")
+                
+        except Exception as e:
+            logging.warning(f"[{self.name}] Memory storage failed: {e}")
+            raise
+        
+        # Only call on_memory_write hook AFTER successful storage
+        if storage_success:
+            try:
+                metadata = kwargs.get('metadata', None)
+                # Try async version first if in async context, fallback to sync
+                import asyncio
+                try:
+                    loop = asyncio.get_running_loop()
+                    # In async context - prefer async hook if available
+                    if hasattr(self._memory_instance, 'aon_memory_write'):
+                        # Schedule async hook without blocking (fire and forget)
+                        asyncio.create_task(self._memory_instance.aon_memory_write(action, memory_type, content, metadata))
+                    elif hasattr(self._memory_instance, 'on_memory_write'):
+                        self._memory_instance.on_memory_write(action, memory_type, content, metadata)
+                except RuntimeError:
+                    # Not in async context - use sync hook
+                    if hasattr(self._memory_instance, 'on_memory_write'):
+                        self._memory_instance.on_memory_write(action, memory_type, content, metadata)
+            except Exception as e:
+                logging.warning(f"[{self.name}] Memory on_memory_write hook failed: {e}")
     
     def _display_memory_info(self):
         """Display memory information to user in a friendly format."""
@@ -4611,60 +4943,57 @@ Answer:"""
                 error=f"Agent guardrail validation error: {str(e)}"
             )
 
-    def _apply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
-        """Apply guardrail validation with retry logic.
-        
-        Args:
-            response_text: The response to validate
-            prompt: Original prompt for regeneration if needed
-            temperature: Temperature for regeneration
-            tools: Tools for regeneration
-            
-        Returns:
-            str: The validated response text or None if validation fails after retries
-        """
+    def _validate_with_guardrail(self, response_text):
+        """Validate response with guardrail. Returns (success, result, error)."""
         if not self._guardrail_fn:
-            return response_text
+            return True, response_text, None
             
         from ..main import TaskOutput
         
+        task_output = TaskOutput(
+            description="Agent response output",
+            raw=response_text,
+            agent=self.name
+        )
+        
+        guardrail_result = self._process_guardrail(task_output)
+        
+        if guardrail_result.success:
+            # Return the potentially modified result
+            if guardrail_result.result and hasattr(guardrail_result.result, 'raw'):
+                return True, guardrail_result.result.raw, None
+            elif guardrail_result.result:
+                return True, str(guardrail_result.result), None
+            else:
+                return True, response_text, None
+        else:
+            return False, None, guardrail_result.error
+
+    def _apply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
+        """Apply guardrail validation with retry logic (sync version)."""
         retry_count = 0
         current_response = response_text
         
         while retry_count <= self.max_guardrail_retries:
-            # Create TaskOutput object
-            task_output = TaskOutput(
-                description="Agent response output",
-                raw=current_response,
-                agent=self.name
-            )
+            success, result, error = self._validate_with_guardrail(current_response)
             
-            # Process guardrail
-            guardrail_result = self._process_guardrail(task_output)
-            
-            if guardrail_result.success:
+            if success:
                 logging.info(f"Agent {self.name}: Guardrail validation passed")
-                # Return the potentially modified result
-                if guardrail_result.result and hasattr(guardrail_result.result, 'raw'):
-                    return guardrail_result.result.raw
-                elif guardrail_result.result:
-                    return str(guardrail_result.result)
-                else:
-                    return current_response
+                return result
             
             # Guardrail failed
             if retry_count >= self.max_guardrail_retries:
                 raise Exception(
                     f"Agent {self.name} response failed guardrail validation after {self.max_guardrail_retries} retries. "
-                    f"Last error: {guardrail_result.error}"
+                    f"Last error: {error}"
                 )
             
             retry_count += 1
-            logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {guardrail_result.error}")
+            logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {error}")
             
             # Regenerate response for retry
             try:
-                retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {guardrail_result.error}. Please provide an improved response."
+                retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {error}. Please provide an improved response."
                 response = self._chat_completion([{"role": "user", "content": retry_prompt}], temperature, tools, task_name=task_name, task_description=task_description, task_id=task_id)
                 if response and response.choices:
                     content = response.choices[0].message.content
@@ -4673,10 +5002,46 @@ Answer:"""
                     raise Exception("Failed to generate retry response")
             except Exception as e:
                 logging.error(f"Agent {self.name}: Error during guardrail retry: {e}")
-                # If we can't regenerate, fail the guardrail
+                raise Exception(f"Agent {self.name} guardrail retry failed: {e}")
+        
+        return current_response
+
+    async def _aapply_guardrail_with_retry(self, response_text, prompt, temperature=1.0, tools=None, task_name=None, task_description=None, task_id=None):
+        """Apply guardrail validation with retry logic (async version)."""
+        retry_count = 0
+        current_response = response_text
+        
+        while retry_count <= self.max_guardrail_retries:
+            success, result, error = self._validate_with_guardrail(current_response)
+            
+            if success:
+                logging.info(f"Agent {self.name}: Guardrail validation passed")
+                return result
+            
+            # Guardrail failed
+            if retry_count >= self.max_guardrail_retries:
                 raise Exception(
-                    f"Agent {self.name} guardrail retry failed: {e}"
+                    f"Agent {self.name} response failed guardrail validation after {self.max_guardrail_retries} retries. "
+                    f"Last error: {error}"
                 )
+            
+            retry_count += 1
+            logging.warning(f"Agent {self.name}: Guardrail validation failed (retry {retry_count}/{self.max_guardrail_retries}): {error}")
+            
+            # Regenerate response for retry (async version)
+            try:
+                retry_prompt = f"{prompt}\n\nNote: Previous response failed validation due to: {error}. Please provide an improved response."
+                response = await self._execute_unified_achat_completion([{"role": "user", "content": retry_prompt}], temperature, tools, task_name=task_name, task_description=task_description, task_id=task_id)
+                if response and hasattr(response, 'choices') and response.choices:
+                    content = response.choices[0].message.content
+                    current_response = content.strip() if content else ""
+                elif isinstance(response, str):
+                    current_response = response.strip()
+                else:
+                    raise Exception("Failed to generate retry response")
+            except Exception as e:
+                logging.error(f"Agent {self.name}: Error during guardrail retry: {e}")
+                raise Exception(f"Agent {self.name} guardrail retry failed: {e}")
         
         return current_response
     

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

@@ -164,7 +164,8 @@ Follow best practices and coding standards."""
             messages=messages,
             **{k: v for k, v in kwargs.items() if k != "model"}
         )
-        
+        if not response.choices or response.choices[0].message is None:
+            raise ValueError("LLM returned empty or filtered response")
         code = response.choices[0].message.content
         self._log("Code generated successfully", "green")
         return code
@@ -204,7 +205,8 @@ Follow best practices and coding standards."""
             messages=messages,
             **{k: v for k, v in kwargs.items() if k != "model"}
         )
-        
+        if not response.choices or response.choices[0].message is None:
+            raise ValueError("LLM returned empty or filtered response")
         code = response.choices[0].message.content
         self._log("Code generated successfully", "green")
         return code
@@ -365,6 +367,8 @@ Provide clear, actionable feedback."""
             **{k: v for k, v in kwargs.items() if k != "model"}
         )
         
+        if not response.choices or response.choices[0].message is None:
+            raise ValueError("LLM returned empty or filtered response")
         review = response.choices[0].message.content
         self._log("Review completed", "green")
         return review
@@ -405,6 +409,8 @@ Include:
             **{k: v for k, v in kwargs.items() if k != "model"}
         )
         
+        if not response.choices or response.choices[0].message is None:
+            raise ValueError("LLM returned empty or filtered response")
         explanation = response.choices[0].message.content
         self._log("Explanation completed", "green")
         return explanation
@@ -451,6 +457,8 @@ Only output the refactored code."""
             **{k: v for k, v in kwargs.items() if k != "model"}
         )
         
+        if not response.choices or response.choices[0].message is None:
+            raise ValueError("LLM returned empty or filtered response")
         refactored = response.choices[0].message.content
         self._log("Refactoring completed", "green")
         return refactored
@@ -495,6 +503,8 @@ Ensure the fix:
             **{k: v for k, v in kwargs.items() if k != "model"}
         )
         
+        if not response.choices or response.choices[0].message is None:
+            raise ValueError("LLM returned empty or filtered response")
         fixed = response.choices[0].message.content
         self._log("Fix completed", "green")
         return fixed

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

@@ -9,6 +9,7 @@ import time
 import json
 import logging
 import inspect
+import os
 
 import asyncio
 import threading
@@ -805,7 +806,25 @@ Write the complete compiled report:"""
                             # Suppress verbose during animation - we'll display result ourselves
                             self.verbose = False
                             current_status[0] = "Sending to LLM..."
-                            result_holder[0] = self.chat(prompt, **kwargs)
+                            
+                            # Check for steering messages before starting chat
+                            if hasattr(self, '_check_steering_messages'):
+                                try:
+                                    steering_msg = self._check_steering_messages()
+                                    if steering_msg:
+                                        # Inject steering message into prompt with clear separator
+                                        prompt_with_steering = f"{prompt}\n\n{steering_msg}"
+                                        current_status[0] = "Processing steering guidance..."
+                                        result_holder[0] = self.chat(prompt_with_steering, **kwargs)
+                                    else:
+                                        result_holder[0] = self.chat(prompt, **kwargs)
+                                except Exception as e:
+                                    logger.warning(f"Steering check failed, continuing without steering: {e}")
+                                    current_status[0] = "Steering failed, proceeding normally..."
+                                    result_holder[0] = self.chat(prompt, **kwargs)
+                            else:
+                                result_holder[0] = self.chat(prompt, **kwargs)
+                                
                             current_status[0] = "Finalizing response..."
                         except Exception as e:
                             error_holder[0] = e
@@ -947,7 +966,80 @@ Write the complete compiled report:"""
         return await self.achat(prompt, task_name=task_name, task_description=task_description, task_id=task_id)
 
     async def execute_tool_async(self, function_name: str, arguments: Dict[str, Any], tool_call_id: Optional[str] = None, tools_override: Optional[List] = None) -> Any:
-        """Async version of execute_tool"""
+        """Async version of execute_tool with retry policy support"""
+        
+        # Get retry policy (tool-level > agent-level > default)
+        retry_policy = self._get_tool_retry_policy(function_name)
+        
+        last_exception = None
+        
+        # Retry loop with exponential backoff
+        for attempt in range(retry_policy.max_attempts):
+            try:
+                result = await self._execute_tool_async_impl(function_name, arguments, tool_call_id, tools_override)
+                
+                # Check if result is an error that should be retried
+                if isinstance(result, dict) and result.get("error"):
+                    # Skip retry for non-retryable errors (approval, permission, etc.)
+                    if (result.get("approval_denied") or 
+                        result.get("permission_denied") or 
+                        result.get("approval_error") or
+                        result.get("circuit_open")):
+                        return result
+                    
+                    # Determine error type for retry policy
+                    error_type = self._classify_error_type(result, last_exception)
+                    
+                    # Check if we should retry
+                    if not retry_policy.should_retry(error_type, attempt):
+                        return result
+                    
+                    # Don't retry on last attempt
+                    if attempt == retry_policy.max_attempts - 1:
+                        return result
+                    
+                    # Emit retry hook event  
+                    delay_ms = retry_policy.get_delay_ms(attempt)
+                    await self._emit_retry_hook_async(function_name, attempt + 1, delay_ms, result.get("error", "Unknown error"), retry_policy.max_attempts, error_type)
+                    
+                    # Wait before retry (async)
+                    await asyncio.sleep(delay_ms / 1000.0)
+                    continue
+                else:
+                    # Success - return result
+                    return result
+                    
+            except Exception as e:
+                last_exception = e
+                # Determine if retryable
+                is_retryable = not isinstance(e, (ValueError, TypeError, AttributeError))
+                
+                # Check if retryable and not last attempt
+                if not is_retryable or attempt == retry_policy.max_attempts - 1:
+                    # Return error dict instead of raising for consistency with sync version
+                    return {"error": f"Error in execute_tool_async: {str(e)}"}
+                
+                # Determine error type  
+                error_type = self._classify_error_type(None, e)
+                
+                if not retry_policy.should_retry(error_type, attempt):
+                    return {"error": f"Error in execute_tool_async: {str(e)}"}
+                
+                # Emit retry hook event
+                delay_ms = retry_policy.get_delay_ms(attempt)
+                await self._emit_retry_hook_async(function_name, attempt + 1, delay_ms, str(e), retry_policy.max_attempts, error_type)
+                
+                # Wait before retry (async)
+                await asyncio.sleep(delay_ms / 1000.0)
+                continue
+        
+        # Should never reach here due to loop logic, but safety fallback
+        if last_exception:
+            return {"error": f"Error in execute_tool_async: {str(last_exception)}"}
+        return {"error": "Maximum retry attempts exceeded"}
+
+    async def _execute_tool_async_impl(self, function_name: str, arguments: Dict[str, Any], tool_call_id: Optional[str] = None, tools_override: Optional[List] = None) -> Any:
+        """Internal async tool execution implementation"""
         try:
             logging.info(f"Executing async tool: {function_name} with arguments: {arguments}")
             
@@ -1286,3 +1378,51 @@ Write the complete compiled report:"""
             print(f"\nTo add MCP capabilities, install: pip install {missing_module}")
             return None 
 
+    async def _emit_retry_hook_async(self, tool_name, attempt, delay_ms, error, max_attempts, error_type):
+        """Emit ON_RETRY hook event (async version).
+        
+        Args:
+            tool_name: Name of the tool being retried
+            attempt: Current attempt number (1-based)
+            delay_ms: Delay before retry in milliseconds
+            error: Error message or description
+            max_attempts: Maximum number of attempts configured
+            error_type: Classified error type
+        """
+        try:
+            from ..hooks import HookEvent, OnRetryInput
+            
+            # Only emit if we have a hook runner
+            hook_runner = getattr(self, '_hook_runner', None)
+            if hook_runner is None:
+                return
+            
+            retry_input = OnRetryInput(
+                session_id=getattr(self, '_session_id', 'default'),
+                cwd=os.getcwd(),
+                event_name=HookEvent.ON_RETRY,
+                timestamp=str(time.time()),
+                agent_name=self.name,
+                tool_name=tool_name,
+                attempt=attempt,
+                delay_ms=delay_ms,
+                error=error,
+                max_attempts=max_attempts,
+                error_type=error_type
+            )
+            
+            # Execute hook asynchronously if supported, otherwise use sync
+            if hasattr(hook_runner, 'execute_async'):
+                await hook_runner.execute_async(HookEvent.ON_RETRY, retry_input, target=tool_name)
+            else:
+                # Fallback to sync execution in executor to avoid blocking event loop
+                loop = asyncio.get_running_loop()
+                await loop.run_in_executor(
+                    None,
+                    lambda: hook_runner.execute_sync(HookEvent.ON_RETRY, retry_input, target=tool_name)
+                )
+            
+        except Exception as e:
+            # Don't let hook failures break retry logic
+            logging.debug(f"Failed to emit async retry hook: {e}")
+

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

@@ -10,7 +10,7 @@ Unified Handoff System:
 - Replaces/absorbs Agent.delegate() and SubagentDelegator functionality
 """
 
-from typing import Optional, Any, Callable, Dict, List, Union, TYPE_CHECKING
+from typing import Optional, Any, Callable, Dict, List, Union, TYPE_CHECKING, Literal, TypeVar, Type, Generic
 from dataclasses import dataclass, field
 from enum import Enum
 import inspect
@@ -19,6 +19,22 @@ from praisonaiagents._logging import get_logger
 import asyncio
 import threading
 import time
+import json
+
+try:
+    from pydantic import BaseModel, ValidationError as PydanticValidationError
+    PYDANTIC_AVAILABLE = True
+except ImportError:
+    PYDANTIC_AVAILABLE = False
+    if TYPE_CHECKING:
+        from pydantic import BaseModel, ValidationError as PydanticValidationError
+    else:
+        # Provide a safe sentinel for runtime
+        class BaseModel:  # type: ignore
+            pass
+        PydanticValidationError = Exception
+
+from ..run_outcome import AgentRunOutcome, RunStatus
 
 if TYPE_CHECKING:
     from .agent import Agent
@@ -32,6 +48,24 @@ class ContextPolicy(Enum):
     NONE = "none"       # No context sharing
     LAST_N = "last_n"   # Share last N messages
 
+@dataclass
+class HandoffToolPolicy:
+    """
+    Policy for tool boundary enforcement during handoff.
+    
+    Defines how tools are filtered when handing off to a target agent.
+    Default mode is 'intersect' for security by default - sub-agents only
+    get tools that both they and the source agent have access to.
+    
+    Attributes:
+        mode: Tool filtering mode
+            - "intersect": Target gets intersection of source and target tools (DEFAULT - secure)
+            - "passthrough": Target keeps its full tool set (legacy behavior - opt-in)
+        blocked_tools: List of tool names to always strip, regardless of intersection
+    """
+    mode: Literal["intersect", "passthrough"] = "intersect"
+    blocked_tools: List[str] = field(default_factory=list)
+
 @dataclass
 class HandoffConfig:
     """
@@ -45,6 +79,7 @@ class HandoffConfig:
         max_context_tokens: Maximum tokens to include in context
         max_context_messages: Maximum messages to include (for LAST_N policy)
         preserve_system: Whether to preserve system messages in context
+        tool_policy: Tool boundary enforcement policy (NEW - security by default)
         timeout_seconds: Timeout for handoff execution
         max_concurrent: Maximum concurrent handoffs (0 = unlimited)
         detect_cycles: Enable cycle detection to prevent infinite loops
@@ -61,6 +96,9 @@ class HandoffConfig:
     max_context_messages: int = 10
     preserve_system: bool = True
     
+    # Tool security boundary (NEW)
+    tool_policy: HandoffToolPolicy = field(default_factory=HandoffToolPolicy)
+    
     # Execution control
     timeout_seconds: float = 300.0
     max_concurrent: int = 5
@@ -85,6 +123,10 @@ class HandoffConfig:
             "max_context_tokens": self.max_context_tokens,
             "max_context_messages": self.max_context_messages,
             "preserve_system": self.preserve_system,
+            "tool_policy": {
+                "mode": self.tool_policy.mode,
+                "blocked_tools": self.tool_policy.blocked_tools.copy()
+            },
             "timeout_seconds": self.timeout_seconds,
             "max_concurrent": self.max_concurrent,
             "detect_cycles": self.detect_cycles,
@@ -96,16 +138,29 @@ class HandoffConfig:
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> 'HandoffConfig':
         """Create config from dictionary."""
-        if "context_policy" in data and isinstance(data["context_policy"], str):
-            data["context_policy"] = ContextPolicy(data["context_policy"])
-        return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
+        data_copy = data.copy()
+        
+        # Handle context_policy enum conversion
+        if "context_policy" in data_copy and isinstance(data_copy["context_policy"], str):
+            data_copy["context_policy"] = ContextPolicy(data_copy["context_policy"])
+        
+        # Handle tool_policy conversion
+        if "tool_policy" in data_copy and isinstance(data_copy["tool_policy"], dict):
+            tool_policy_data = data_copy["tool_policy"]
+            data_copy["tool_policy"] = HandoffToolPolicy(
+                mode=tool_policy_data.get("mode", "intersect"),
+                blocked_tools=tool_policy_data.get("blocked_tools", [])
+            )
+        
+        return cls(**{k: v for k, v in data_copy.items() if k in cls.__dataclass_fields__})
 
 # Import structured error hierarchy from central errors module
 from ..errors import (
     HandoffError, 
     HandoffCycleError, 
     HandoffDepthError, 
-    HandoffTimeoutError
+    HandoffTimeoutError,
+    HandoffValidationError
 )
 
 # Thread-local storage for tracking handoff chains
@@ -147,7 +202,12 @@ class HandoffInputData:
 
 @dataclass 
 class HandoffResult:
-    """Result of a handoff operation."""
+    """
+    Result of a handoff operation.
+    
+    Now includes typed outcome information while maintaining backward
+    compatibility with the legacy boolean success field.
+    """
     success: bool
     response: Optional[str] = None
     target_agent: Optional[str] = None
@@ -156,6 +216,64 @@ class HandoffResult:
     error: Optional[str] = None
     handoff_depth: int = 0
     
+    # New typed outcome field
+    outcome: Optional[AgentRunOutcome] = None
+    
+    def __post_init__(self):
+        """Initialize outcome field if not provided."""
+        if self.outcome is None:
+            # Create outcome based on legacy success field
+            if self.success:
+                self.outcome = AgentRunOutcome.success(
+                    output=self.response or "",
+                    elapsed_s=self.duration_seconds,
+                    agent_name=self.target_agent,
+                    context={
+                        "source_agent": self.source_agent,
+                        "handoff_depth": self.handoff_depth,
+                    }
+                )
+            else:
+                error_text = (self.error or "").lower()
+                context = {
+                    "source_agent": self.source_agent,
+                    "handoff_depth": self.handoff_depth,
+                }
+                if any(keyword in error_text for keyword in ["timeout", "timed out"]):
+                    self.outcome = AgentRunOutcome.timeout(
+                        error=self.error or "Handoff failed",
+                        elapsed_s=self.duration_seconds,
+                        agent_name=self.target_agent,
+                        context=context,
+                    )
+                else:
+                    self.outcome = AgentRunOutcome.failure(
+                        error=self.error or "Handoff failed",
+                        elapsed_s=self.duration_seconds,
+                        agent_name=self.target_agent,
+                        context=context,
+                    )
+    
+    @classmethod
+    def from_outcome(
+        cls,
+        outcome: AgentRunOutcome,
+        target_agent: Optional[str] = None,
+        source_agent: Optional[str] = None,
+        handoff_depth: int = 0,
+    ) -> "HandoffResult":
+        """Create HandoffResult from AgentRunOutcome."""
+        return cls(
+            success=outcome.is_success(),
+            response=outcome.output,
+            target_agent=target_agent or outcome.agent_name,
+            source_agent=source_agent,
+            duration_seconds=outcome.elapsed_s,
+            error=outcome.error,
+            handoff_depth=handoff_depth,
+            outcome=outcome,
+        )
+    
     
 class Handoff:
     """
@@ -241,6 +359,52 @@ class Handoff:
             agent_desc += f" - {self.agent.goal}"
         return agent_desc
         
+    def _compute_effective_tools(self, source_agent: 'Agent') -> List[Any] | None:
+        """
+        Compute the effective tool set for the target agent based on tool policy.
+        
+        Args:
+            source_agent: The agent initiating the handoff
+            
+        Returns:
+            List of tools the target agent should have access to during handoff.
+            Returns None to indicate unrestricted access (for passthrough mode without blocked tools).
+            Returns [] to indicate no tools allowed (for intersect mode with empty intersection).
+        """
+        policy = self.config.tool_policy
+        
+        if policy.mode == "passthrough":
+            # Legacy behavior - target keeps its full tool set, minus blocked tools
+            if not policy.blocked_tools:
+                # No restrictions - return None so chat() uses the agent's configured tools
+                return None
+            else:
+                # Filter out blocked tools
+                target_tools = getattr(self.agent, 'tools', None) or []
+                effective_tools = [
+                    tool for tool in target_tools
+                    if getattr(tool, 'name', getattr(tool, '__name__', str(tool))) not in policy.blocked_tools
+                ]
+                return effective_tools
+        else:  # intersect mode (default)
+            # Security by default - intersection of source and target tools
+            source_tools = getattr(source_agent, 'tools', None) or []
+            source_tool_names = {
+                getattr(tool, 'name', getattr(tool, '__name__', str(tool)))
+                for tool in source_tools
+            }
+            
+            target_tools = getattr(self.agent, 'tools', None) or []
+            effective_tools = []
+            for tool in target_tools:
+                tool_name = getattr(tool, 'name', getattr(tool, '__name__', str(tool)))
+                if (tool_name in source_tool_names and 
+                    tool_name not in policy.blocked_tools):
+                    effective_tools.append(tool)
+            
+            # In intersect mode, empty list means no tools allowed - this is the security boundary
+            return effective_tools
+    
     def _check_safety(self, source_agent: 'Agent') -> None:
         """
         Check safety constraints before handoff.
@@ -402,8 +566,11 @@ class Handoff:
             
             logger.info(f"Programmatic handoff to {self.agent.name}")
             
-            # Execute with timeout if sync
-            response = self.agent.chat(full_prompt)
+            # Compute effective tools based on tool policy
+            effective_tools = self._compute_effective_tools(source_agent)
+            
+            # Execute with tool boundary enforcement
+            response = self.agent.chat(full_prompt, tools=effective_tools)
             
             result = HandoffResult(
                 success=True,
@@ -417,6 +584,20 @@ class Handoff:
             # Execute on_complete callback
             self._execute_callback(self.config.on_complete, source_agent, kwargs, result)
             
+            # Call on_delegation hook on source agent's memory
+            if (hasattr(source_agent, '_memory_instance') and 
+                source_agent._memory_instance and
+                hasattr(source_agent._memory_instance, 'on_delegation')):
+                try:
+                    source_agent._memory_instance.on_delegation(
+                        task=full_prompt,
+                        result=result.response or "",
+                        agent_name=self.agent.name,
+                        metadata={"handoff_depth": result.handoff_depth}
+                    )
+                except Exception as e:
+                    logger.warning(f"[{source_agent.name}] Memory on_delegation hook failed: {e}")
+            
             return result
             
         except (HandoffCycleError, HandoffDepthError, HandoffTimeoutError) as e:
@@ -491,15 +672,18 @@ class Handoff:
                 
                 logger.info(f"Async handoff to {self.agent.name}")
                 
-                # Execute - check for async chat method
-                if hasattr(self.agent, 'achat'):
-                    response = await self.agent.achat(full_prompt)
-                else:
-                    # Run sync chat in executor
-                    loop = asyncio.get_event_loop()
-                    response = await loop.run_in_executor(None, self.agent.chat, full_prompt)
+                # Compute effective tools based on tool policy
+                effective_tools = self._compute_effective_tools(source_agent)
                 
-                return HandoffResult(
+                # Execute with tool boundary enforcement - check for async chat method
+                if hasattr(self.agent, 'achat'):
+                    response = await self.agent.achat(full_prompt, tools=effective_tools)
+                else:
+                    # Run sync chat in executor with tool constraint
+                    loop = asyncio.get_event_loop()
+                    response = await loop.run_in_executor(None, lambda: self.agent.chat(full_prompt, tools=effective_tools))
+                
+                result = HandoffResult(
                     success=True,
                     response=str(response) if response else "",
                     target_agent=self.agent.name,
@@ -507,6 +691,22 @@ class Handoff:
                     duration_seconds=time.time() - start_time,
                     handoff_depth=_get_handoff_depth(),
                 )
+                
+                # Call on_delegation hook on source agent's memory
+                if (hasattr(source_agent, '_memory_instance') and 
+                    source_agent._memory_instance and
+                    hasattr(source_agent._memory_instance, 'on_delegation')):
+                    try:
+                        source_agent._memory_instance.on_delegation(
+                            task=full_prompt,
+                            result=result.response or "",
+                            agent_name=self.agent.name,
+                            metadata={"handoff_depth": result.handoff_depth}
+                        )
+                    except Exception as e:
+                        logger.warning(f"[{source_agent.name}] Memory on_delegation hook failed: {e}")
+                
+                return result
             finally:
                 _pop_handoff()
         
@@ -609,11 +809,15 @@ class Handoff:
                 if kwargs and self.input_type:
                     context_info += f"Context: {kwargs} "
                 
-                # Execute the target agent
+                # Execute the target agent with tool boundary enforcement
                 if last_message:
                     prompt = context_info + last_message
                     logger.info(f"Handing off to {self.agent.name} with prompt: {prompt}")
-                    response = self.agent.chat(prompt)
+                    
+                    # Compute effective tools based on tool policy
+                    effective_tools = self._compute_effective_tools(source_agent)
+                    
+                    response = self.agent.chat(prompt, tools=effective_tools)
                     
                     result = HandoffResult(
                         success=True,
@@ -625,6 +829,20 @@ class Handoff:
                     )
                     self._execute_callback(self.config.on_complete, source_agent, kwargs, result)
                     
+                    # Call on_delegation hook on source agent's memory
+                    if (hasattr(source_agent, '_memory_instance') and 
+                        source_agent._memory_instance and
+                        hasattr(source_agent._memory_instance, 'on_delegation')):
+                        try:
+                            source_agent._memory_instance.on_delegation(
+                                task=prompt,
+                                result=result.response or "",
+                                agent_name=self.agent.name,
+                                metadata={"handoff_depth": result.handoff_depth}
+                            )
+                        except Exception as e:
+                            logger.warning(f"[{source_agent.name}] Memory on_delegation hook failed: {e}")
+                    
                     return f"Handoff successful. {self.agent.name} response: {response}"
                 
                 return f"Handoff to {self.agent.name} completed, but no specific task was provided."
@@ -682,13 +900,16 @@ def handoff(
     max_concurrent: Optional[int] = None,
     detect_cycles: Optional[bool] = None,
     max_depth: Optional[int] = None,
+    # Tool security boundary kwargs (NEW)
+    tool_policy_mode: Optional[Literal["intersect", "passthrough"]] = None,
+    blocked_tools: Optional[List[str]] = None,
 ) -> Handoff:
     """
     Create a handoff configuration for delegating tasks to another agent.
     
     This is a convenience function that creates a Handoff instance with the
     specified configuration. It supports both the legacy API and the new
-    unified HandoffConfig.
+    unified HandoffConfig with tool security boundary enforcement.
     
     Args:
         agent: The target agent to hand off to
@@ -703,29 +924,39 @@ def handoff(
         max_concurrent: Shorthand for config.max_concurrent
         detect_cycles: Shorthand for config.detect_cycles
         max_depth: Shorthand for config.max_depth
+        tool_policy_mode: Tool boundary mode ("intersect" = secure by default, "passthrough" = legacy)
+        blocked_tools: List of tool names to always block regardless of intersection
         
     Returns:
         A configured Handoff instance
         
     Example:
         ```python
-        from praisonaiagents import Agent, handoff, HandoffConfig
+        from praisonaiagents import Agent, handoff, HandoffConfig, HandoffToolPolicy
         
         billing_agent = Agent(name="Billing Agent")
-        refund_agent = Agent(name="Refund Agent")
+        refund_agent = Agent(name="Refund Agent") 
         
-        # Simple usage
+        # Simple usage (secure by default - tools intersected)
         triage_agent = Agent(
             name="Triage Agent",
             handoffs=[billing_agent, handoff(refund_agent)]
         )
         
-        # With config
+        # With tool security config
         triage_agent = Agent(
-            name="Triage Agent",
+            name="Triage Agent", 
             handoffs=[
-                handoff(billing_agent, context_policy="summary", timeout_seconds=60),
-                handoff(refund_agent, config=HandoffConfig(detect_cycles=True))
+                handoff(billing_agent, 
+                        tool_policy_mode="intersect",  # Only shared tools
+                        blocked_tools=["execute_code", "shell_tools"]),
+                handoff(refund_agent,
+                        config=HandoffConfig(
+                            tool_policy=HandoffToolPolicy(
+                                mode="passthrough",  # Legacy behavior
+                                blocked_tools=["dangerous_tool"]
+                            )
+                        ))
             ]
         )
         ```
@@ -746,6 +977,14 @@ def handoff(
     if max_depth is not None:
         config.max_depth = max_depth
     
+    # Apply tool policy kwargs to config
+    if tool_policy_mode is not None or blocked_tools is not None:
+        # Create new tool policy with updated values
+        config.tool_policy = HandoffToolPolicy(
+            mode=tool_policy_mode if tool_policy_mode is not None else config.tool_policy.mode,
+            blocked_tools=blocked_tools if blocked_tools is not None else config.tool_policy.blocked_tools
+        )
+    
     return Handoff(
         agent=agent,
         tool_name_override=tool_name_override,
@@ -924,6 +1163,402 @@ async def parallel_handoffs(
     return processed_results
 
 
+# Type variable for generic typed handoff
+T = TypeVar("T", bound=BaseModel)
+
+
+class TypedHandoff(Handoff, Generic[T]):
+    """
+    Typed handoff with schema validation using Pydantic models.
+    
+    This class extends the base Handoff to add type safety through Pydantic schema validation.
+    The sender declares an output schema, the receiver expects that schema, and the framework
+    validates the payload at the boundary, raising HandoffValidationError if validation fails.
+    
+    The receiving agent's prompt is constructed from validated JSON rather than raw string
+    concatenation, enabling proper deserialization of structured data.
+    
+    Example:
+        ```python
+        from pydantic import BaseModel
+        from praisonaiagents.agent.handoff import TypedHandoff, HandoffValidationError
+        
+        class ResearchResult(BaseModel):
+            summary: str
+            citations: list[str] 
+            confidence: float
+            
+        # Create typed handoff
+        typed_handoff = TypedHandoff(
+            agent=writer_agent,
+            input_schema=ResearchResult
+        )
+        
+        # Valid payload
+        result = ResearchResult(
+            summary="AI research findings", 
+            citations=["ref1", "ref2"], 
+            confidence=0.92
+        )
+        response = typed_handoff.execute_programmatic(source_agent, result)
+        
+        # Invalid payload raises HandoffValidationError
+        bad_payload = {"summary": "...", "citations": "not-a-list"}
+        typed_handoff.execute_programmatic(source_agent, bad_payload)  # Raises error
+        ```
+    """
+    
+    def __init__(
+        self,
+        agent: 'Agent',
+        input_schema: Type[T],
+        tool_name_override: Optional[str] = None,
+        tool_description_override: Optional[str] = None,
+        on_handoff: Optional[Callable] = None,
+        input_filter: Optional[Union[Callable[[HandoffInputData], HandoffInputData], List[Callable[[HandoffInputData], HandoffInputData]]]] = None,
+        config: Optional[HandoffConfig] = None,
+    ):
+        """
+        Initialize a TypedHandoff with schema validation.
+        
+        Args:
+            agent: The target agent to hand off to
+            input_schema: Pydantic model class for payload validation
+            tool_name_override: Custom tool name (defaults to transfer_to_<agent_name>)
+            tool_description_override: Custom tool description
+            on_handoff: Callback function executed when handoff is invoked
+            input_filter: Function or list of functions to filter/transform input
+            config: HandoffConfig for advanced settings (context policy, timeouts, etc.)
+            
+        Raises:
+            ImportError: If Pydantic is not available
+            TypeError: If input_schema is not a Pydantic model
+        """
+        if not PYDANTIC_AVAILABLE:
+            raise ImportError(
+                "Pydantic is required for TypedHandoff. Install with: pip install pydantic"
+            )
+            
+        if not (inspect.isclass(input_schema) and issubclass(input_schema, BaseModel)):
+            raise TypeError(
+                f"input_schema must be a Pydantic BaseModel class, got {type(input_schema)}"
+            )
+        
+        super().__init__(
+            agent=agent,
+            tool_name_override=tool_name_override,
+            tool_description_override=tool_description_override,
+            on_handoff=on_handoff,
+            input_type=None,  # We handle typing ourselves
+            input_filter=input_filter,
+            config=config,
+        )
+        self._input_schema = input_schema
+    
+    def _validate_payload(self, payload: Union[T, dict, Any]) -> T:
+        """
+        Validate payload against the input schema.
+        
+        Args:
+            payload: The payload to validate - can be a Pydantic model instance,
+                    dict, or any object with model_dump() method
+                    
+        Returns:
+            Validated Pydantic model instance
+            
+        Raises:
+            HandoffValidationError: If validation fails
+        """
+        try:
+            if isinstance(payload, self._input_schema):
+                # Already the correct type, validate by re-parsing
+                validated = self._input_schema.model_validate(payload.model_dump())
+            elif isinstance(payload, dict):
+                # Dictionary - validate directly
+                validated = self._input_schema.model_validate(payload)
+            elif hasattr(payload, 'model_dump'):
+                # Another Pydantic model - convert via dict
+                validated = self._input_schema.model_validate(payload.model_dump())
+            else:
+                # Try to convert to dict first
+                if hasattr(payload, '__dict__'):
+                    payload_dict = payload.__dict__
+                else:
+                    payload_dict = dict(payload) if hasattr(payload, '__iter__') else {}
+                validated = self._input_schema.model_validate(payload_dict)
+                
+            return validated
+            
+        except PydanticValidationError as exc:
+            validation_errors = [
+                f"{err['loc']}: {err['msg']}" for err in exc.errors()
+            ] if hasattr(exc, 'errors') else [str(exc)]
+            
+            raise HandoffValidationError(
+                f"Handoff to '{self.agent.name}' received an invalid payload: {exc}",
+                source_agent="unknown",
+                target_agent=self.agent.name,
+                agent_id="unknown",
+                validation_errors=validation_errors
+            ) from exc
+        except Exception as exc:
+            raise HandoffValidationError(
+                f"Handoff to '{self.agent.name}' failed to validate payload: {exc}",
+                source_agent="unknown", 
+                target_agent=self.agent.name,
+                agent_id="unknown"
+            ) from exc
+    
+    def execute_programmatic(
+        self,
+        source_agent: 'Agent',
+        payload: Union[T, dict, str],
+        context: Optional[Dict[str, Any]] = None,
+    ) -> HandoffResult:
+        """
+        Execute typed handoff programmatically with schema validation.
+        
+        Args:
+            source_agent: The agent initiating the handoff
+            payload: The typed payload (Pydantic model instance or dict) OR
+                    a string prompt (for backward compatibility)
+            context: Optional additional context
+            
+        Returns:
+            HandoffResult with response or error
+            
+        Raises:
+            HandoffValidationError: If payload validation fails
+        """
+        # Handle backward compatibility: if payload is a string, treat as regular handoff
+        if isinstance(payload, str):
+            return super().execute_programmatic(source_agent, payload, context)
+        
+        start_time = time.time()
+        kwargs = context or {}
+        
+        try:
+            # Safety checks
+            self._check_safety(source_agent)
+            
+            # Track handoff chain
+            _push_handoff(source_agent.name)
+            
+            # Execute on_handoff callback
+            self._execute_callback(self.config.on_handoff or self.on_handoff, source_agent, kwargs)
+            
+            # Apply input filter if provided (fixes Greptile P1 issue)
+            _ = self._prepare_context(source_agent, kwargs)
+            
+            # Validate payload against schema
+            try:
+                validated_payload = self._validate_payload(payload)
+            except HandoffValidationError as e:
+                # Update error context with proper agent info (fixes Greptile P2 issue)
+                e.source_agent = source_agent.name
+                e.agent_id = source_agent.name
+                e.context["source_agent"] = source_agent.name
+                e.context["agent_id"] = source_agent.name
+                raise
+            
+            # Convert validated payload to structured JSON for the prompt
+            payload_json = validated_payload.model_dump_json(indent=2)
+            
+            # Build prompt with structured JSON instead of string concatenation
+            context_prefix = f"[Handoff from {source_agent.name}] "
+            if kwargs:
+                context_prefix += f"Additional context: {json.dumps(kwargs, indent=2)}\n"
+            
+            # The key improvement: structured JSON instead of str() concatenation
+            full_prompt = f"{context_prefix}Payload:\n{payload_json}"
+            
+            logger.info(f"Typed handoff to {self.agent.name} with validated payload")
+            
+            # Execute with timeout if sync
+            response = self.agent.chat(full_prompt)
+            
+            result = HandoffResult(
+                success=True,
+                response=str(response) if response else "",
+                target_agent=self.agent.name,
+                source_agent=source_agent.name,
+                duration_seconds=time.time() - start_time,
+                handoff_depth=_get_handoff_depth(),
+            )
+            
+            # Execute on_complete callback
+            self._execute_callback(self.config.on_complete, source_agent, kwargs, result)
+            
+            return result
+            
+        except (HandoffCycleError, HandoffDepthError, HandoffTimeoutError, HandoffValidationError) as e:
+            result = HandoffResult(
+                success=False,
+                target_agent=self.agent.name,
+                source_agent=source_agent.name,
+                duration_seconds=time.time() - start_time,
+                error=str(e),
+                handoff_depth=_get_handoff_depth(),
+            )
+            self._execute_callback(self.config.on_error, source_agent, kwargs, result)
+            raise
+        except Exception as e:
+            result = HandoffResult(
+                success=False,
+                target_agent=self.agent.name,
+                source_agent=source_agent.name,
+                duration_seconds=time.time() - start_time,
+                error=str(e),
+                handoff_depth=_get_handoff_depth(),
+            )
+            self._execute_callback(self.config.on_error, source_agent, kwargs, result)
+            logger.error(f"Typed handoff error: {e}")
+            return result
+        finally:
+            _pop_handoff()
+    
+    async def execute_async(
+        self,
+        source_agent: 'Agent',
+        payload: Union[T, dict, str],
+        context: Optional[Dict[str, Any]] = None,
+    ) -> HandoffResult:
+        """
+        Execute typed handoff asynchronously with schema validation.
+        
+        Args:
+            source_agent: The agent initiating the handoff
+            payload: The typed payload (Pydantic model instance or dict) OR
+                    a string prompt (for backward compatibility)
+            context: Optional additional context
+            
+        Returns:
+            HandoffResult with response or error
+            
+        Raises:
+            HandoffValidationError: If payload validation fails
+        """
+        # Handle backward compatibility: if payload is a string, treat as regular handoff
+        if isinstance(payload, str):
+            return await super().execute_async(source_agent, payload, context)
+        
+        # Initialize semaphore if needed (thread-safe)
+        if self.config.max_concurrent > 0 and Handoff._semaphore is None:
+            with Handoff._semaphore_lock:
+                if Handoff._semaphore is None:
+                    Handoff._semaphore = asyncio.Semaphore(self.config.max_concurrent)
+        
+        start_time = time.time()
+        kwargs = context or {}
+        
+        async def _execute():
+            try:
+                # Safety checks
+                self._check_safety(source_agent)
+                _push_handoff(source_agent.name)
+                
+                # Execute callback
+                self._execute_callback(self.config.on_handoff or self.on_handoff, source_agent, kwargs)
+                
+                # Apply input filter if provided (fixes Greptile P1 issue)
+                _ = self._prepare_context(source_agent, kwargs)
+                
+                # Validate payload against schema
+                try:
+                    validated_payload = self._validate_payload(payload)
+                except HandoffValidationError as e:
+                    # Update error context with proper agent info (fixes Greptile P2 issue)
+                    e.source_agent = source_agent.name
+                    e.agent_id = source_agent.name
+                    e.context["source_agent"] = source_agent.name
+                    e.context["agent_id"] = source_agent.name
+                    raise
+                
+                # Convert to structured JSON
+                payload_json = validated_payload.model_dump_json(indent=2)
+                
+                # Build prompt
+                context_prefix = f"[Handoff from {source_agent.name}] "
+                if kwargs:
+                    context_prefix += f"Additional context: {json.dumps(kwargs, indent=2)}\n"
+                
+                full_prompt = f"{context_prefix}Payload:\n{payload_json}"
+                
+                logger.info(f"Async typed handoff to {self.agent.name}")
+                
+                # Execute - check for async chat method
+                if hasattr(self.agent, 'achat'):
+                    response = await self.agent.achat(full_prompt)
+                else:
+                    # Run sync chat in executor
+                    loop = asyncio.get_event_loop()
+                    response = await loop.run_in_executor(None, self.agent.chat, full_prompt)
+                
+                return HandoffResult(
+                    success=True,
+                    response=str(response) if response else "",
+                    target_agent=self.agent.name,
+                    source_agent=source_agent.name,
+                    duration_seconds=time.time() - start_time,
+                    handoff_depth=_get_handoff_depth(),
+                )
+            finally:
+                _pop_handoff()
+        
+        try:
+            if self.config.max_concurrent > 0 and Handoff._semaphore:
+                async with Handoff._semaphore:
+                    if self.config.timeout_seconds > 0:
+                        result = await asyncio.wait_for(
+                            _execute(),
+                            timeout=self.config.timeout_seconds
+                        )
+                    else:
+                        result = await _execute()
+            else:
+                if self.config.timeout_seconds > 0:
+                    result = await asyncio.wait_for(
+                        _execute(),
+                        timeout=self.config.timeout_seconds
+                    )
+                else:
+                    result = await _execute()
+            
+            self._execute_callback(self.config.on_complete, source_agent, kwargs, result)
+            return result
+            
+        except asyncio.TimeoutError as err:
+            result = HandoffResult(
+                success=False,
+                target_agent=self.agent.name,
+                source_agent=source_agent.name,
+                duration_seconds=time.time() - start_time,
+                error=f"Timeout after {self.config.timeout_seconds}s",
+                handoff_depth=_get_handoff_depth(),
+            )
+            self._execute_callback(self.config.on_error, source_agent, kwargs, result)
+            raise HandoffTimeoutError(
+                f"Typed handoff to {self.agent.name} timed out after {self.config.timeout_seconds}s",
+                timeout_seconds=self.config.timeout_seconds,
+                source_agent=source_agent.name,
+                target_agent=self.agent.name,
+                agent_id=source_agent.name
+            ) from err
+        except Exception as e:
+            result = HandoffResult(
+                success=False,
+                target_agent=self.agent.name,
+                source_agent=source_agent.name,
+                duration_seconds=time.time() - start_time,
+                error=str(e),
+                handoff_depth=_get_handoff_depth(),
+            )
+            self._execute_callback(self.config.on_error, source_agent, kwargs, result)
+            if isinstance(e, (HandoffCycleError, HandoffDepthError, HandoffValidationError)):
+                raise
+            return result
+
+
 def prompt_with_handoff_instructions(base_prompt: str, agent: 'Agent') -> str:
     """
     Add handoff instructions to an agent's prompt.

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

@@ -403,9 +403,39 @@ class MemoryMixin:
                     self._session_store.add_user_message(self._session_id, content)
                 elif role == "assistant":
                     self._session_store.add_assistant_message(self._session_id, content)
+                # Keep auto_save index in sync when per-turn persist shares session_id
+                if self.auto_save and self.auto_save == self._session_id:
+                    with self._history_lock:
+                        self._auto_save_last_index = (
+                            getattr(self, "_auto_save_last_index", 0) + 1
+                        )
+                self._persist_session_stats()
             except Exception as e:
                 logging.warning(f"Failed to persist message to session store: {e}")
 
+    def _persist_session_stats(self):
+        """Flush agent cost/token stats into session JSON metadata."""
+        store = getattr(self, "_session_store", None)
+        session_id = getattr(self, "_session_id", None)
+        if store is None or session_id is None:
+            return
+        if not hasattr(store, "update_session_metadata"):
+            return
+        try:
+            summary = getattr(self, "cost_summary", {}) or {}
+            llm = getattr(self, "llm", None)
+            model = str(llm) if llm is not None else None
+            store.update_session_metadata(
+                session_id,
+                model=model,
+                total_tokens=summary.get("tokens_in", 0) + summary.get("tokens_out", 0),
+                cost=summary.get("cost", 0),
+                source=getattr(self, "_session_source", "chat"),
+                agent_id=getattr(self, "agent_id", None) or getattr(self, "_agent_id", None),
+            )
+        except Exception as e:
+            logging.debug(f"Session stats persist skipped: {e}")
+
     def session_id(self) -> Optional[str]:
         """Get the current session ID."""
         return self._session_id

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

@@ -0,0 +1,273 @@
+"""
+Message Steering Implementation.
+
+Core implementation of real-time message steering during agent execution.
+Integrates with existing message queue infrastructure.
+"""
+import time
+import logging
+from typing import List, Optional, Dict, Any
+from .message_queue import AgentMessageQueue, MessagePriority
+from .protocols import (
+    MessageSteeringProtocol, 
+    SteeringMessage, 
+    SteeringPriority,
+    AgentSteeringProtocol
+)
+
+logger = logging.getLogger(__name__)
+
+
+class MessageSteering:
+    """
+    Core message steering implementation.
+    
+    Provides real-time message queuing and processing for agents
+    during execution. Uses the existing AgentMessageQueue infrastructure.
+    """
+    
+    def __init__(self, max_messages: int = 50):
+        """
+        Initialize message steering.
+        
+        Args:
+            max_messages: Maximum number of pending messages
+        """
+        self._message_queue = AgentMessageQueue(max_size=max_messages)
+        self._enabled = True
+        self._last_check = 0.0
+        self._check_interval = 0.1  # Check every 100ms
+        
+    def queue_message(self, message: str, priority: int = 5) -> str:
+        """Queue a steering message."""
+        if not self._enabled:
+            return ""
+            
+        # Convert priority to SteeringPriority
+        try:
+            steering_priority = SteeringPriority(priority)
+        except ValueError:
+            steering_priority = SteeringPriority.NORMAL
+            
+        steering_msg = SteeringMessage(
+            content=message,
+            priority=steering_priority,
+            metadata={"timestamp": time.time()}
+        )
+        
+        # Use existing message queue with priority mapping
+        # Special handling for INTERRUPT - give it maximum priority
+        if steering_priority == SteeringPriority.INTERRUPT:
+            queue_priority = MessagePriority.URGENT.value + 1  # Higher than URGENT
+        else:
+            queue_priority = min(priority, MessagePriority.URGENT.value)
+        success = self._message_queue.enqueue(
+            content=steering_msg,
+            priority=queue_priority,
+            metadata={"type": "steering", "original_priority": priority}
+        )
+        
+        if success:
+            msg_id = f"steer_{int(time.time() * 1000)}"
+            logger.debug(f"Queued steering message: {msg_id}")
+            return msg_id
+        else:
+            logger.warning("Failed to queue steering message - queue full")
+            return ""
+    
+    def get_pending_messages(self) -> List[SteeringMessage]:
+        """Get all pending steering messages."""
+        if not self._enabled:
+            return []
+            
+        messages = []
+        all_messages = self._message_queue.get_all()
+        
+        for msg in all_messages:
+            if isinstance(msg, SteeringMessage):
+                messages.append(msg)
+                
+        return messages
+    
+    def process_steering(self, context: Optional[Dict[str, Any]] = None) -> bool:
+        """
+        Process pending steering messages.
+        
+        This is called during agent execution to check for and process
+        any steering messages. Uses rate limiting to avoid excessive checking,
+        but allows INTERRUPT priority messages to bypass rate limiting.
+        
+        Args:
+            context: Execution context that can be updated with steering info
+            
+        Returns:
+            True if messages were processed, False otherwise
+        """
+        if not self._enabled:
+            return False
+            
+        # Check if we have high priority messages that bypass rate limiting
+        has_urgent_messages = False
+        all_messages = self._message_queue.get_all()
+        for msg in all_messages:
+            if isinstance(msg, SteeringMessage) and msg.priority.value >= SteeringPriority.HIGH.value:
+                has_urgent_messages = True
+                break
+        
+        # Rate limiting - only check every 100ms unless we have urgent messages
+        current_time = time.time()
+        if not has_urgent_messages and current_time - self._last_check < self._check_interval:
+            return False
+            
+        self._last_check = current_time
+        
+        if self._message_queue.is_empty():
+            return False
+            
+        # Process one message per check to avoid blocking
+        msg_content = self._message_queue.dequeue(timeout=0)
+        if msg_content is None:
+            return False
+            
+        if not isinstance(msg_content, SteeringMessage):
+            return False
+            
+        # Update context with steering message
+        if context is not None:
+            if "steering_messages" not in context:
+                context["steering_messages"] = []
+            context["steering_messages"].append({
+                "content": msg_content.content,
+                "priority": msg_content.priority.name,
+                "timestamp": msg_content.metadata.get("timestamp") if msg_content.metadata else None
+            })
+            
+        logger.info(f"Processed steering message: {msg_content.content[:50]}...")
+        return True
+    
+    def clear_messages(self) -> int:
+        """Clear all pending messages."""
+        return self._message_queue.clear()
+    
+    def has_pending_messages(self) -> bool:
+        """Check if there are pending steering messages."""
+        return not self._message_queue.is_empty()
+    
+    def enable(self) -> None:
+        """Enable message steering."""
+        self._enabled = True
+        
+    def disable(self) -> None:
+        """Disable message steering."""
+        self._enabled = False
+        
+    @property
+    def enabled(self) -> bool:
+        """Whether message steering is enabled."""
+        return self._enabled
+
+
+class SteeringMixin:
+    """
+    Mixin to add message steering capabilities to Agent class.
+    
+    This provides the agent interface for message steering while
+    delegating implementation to MessageSteering.
+    """
+    
+    def _init_message_steering(self, message_steering=False):
+        """Initialize message steering if enabled."""
+        if message_steering is True:
+            # Use default implementation
+            self._message_steering = MessageSteering()
+        elif message_steering is False or message_steering is None:
+            # Disabled
+            self._message_steering = None
+        else:
+            # Custom implementation provided
+            self._message_steering = message_steering
+    
+    def steer(self, message: str, priority: int = 5) -> str:
+        """
+        Send a steering message to the agent during execution.
+        
+        This can be called while the agent is running to provide
+        real-time guidance or course corrections.
+        
+        Args:
+            message: The steering message
+            priority: Message priority (1=low, 5=normal, 10=high, 20=urgent, 30=interrupt)
+            
+        Returns:
+            Message ID for tracking, empty string if steering disabled
+            
+        Example:
+            ```python
+            agent = Agent(name="research", message_steering=True)
+            
+            # Start long-running task
+            import threading
+            def run_task():
+                return agent.start("Research AI trends comprehensively")
+            thread = threading.Thread(target=run_task)
+            thread.start()
+            
+            # Send steering messages while running
+            agent.steer("Focus on the business impact, not technical details")
+            agent.steer("Also include information about market size", priority=10)
+            ```
+        """
+        if self._message_steering is None:
+            logger.warning("Message steering not enabled - call ignored")
+            return ""
+            
+        return self._message_steering.queue_message(message, priority)
+    
+    def get_steering_status(self) -> Dict[str, Any]:
+        """Get current steering status."""
+        if self._message_steering is None:
+            return {"enabled": False, "pending_count": 0}
+            
+        return {
+            "enabled": self._message_steering.enabled,
+            "pending_count": len(self._message_steering.get_pending_messages()),
+            "has_pending": self._message_steering.has_pending_messages()
+        }
+    
+    @property
+    def message_steering_enabled(self) -> bool:
+        """Whether message steering is enabled for this agent."""
+        return self._message_steering is not None and self._message_steering.enabled
+    
+    def _check_steering_messages(self) -> Optional[str]:
+        """
+        Check for and process steering messages.
+        
+        This is called during agent execution loops to process
+        any pending steering messages.
+        
+        Returns:
+            Steering message to inject into conversation, or None
+        """
+        if self._message_steering is None:
+            return None
+            
+        # Process steering messages
+        context = {}
+        if self._message_steering.process_steering(context):
+            steering_messages = context.get("steering_messages", [])
+            if steering_messages:
+                # Get the most recent message
+                latest = steering_messages[-1]
+                content = latest["content"]
+                priority = latest["priority"]
+                
+                # Format as system message
+                if priority == "INTERRUPT":
+                    return f"\n[INTERRUPT USER GUIDANCE]: {content}\nPlease stop current work and follow this guidance immediately."
+                elif priority in ("HIGH", "URGENT"):
+                    return f"\n[URGENT USER GUIDANCE]: {content}\nPlease acknowledge and adjust your approach accordingly."
+                else:
+                    return f"\n[USER GUIDANCE]: {content}\nPlease consider this feedback as you continue."
+        
+        return None

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

@@ -10,6 +10,25 @@ This enables:
 These protocols are lightweight and have zero performance impact.
 """
 from typing import Protocol, runtime_checkable, Optional, Any, AsyncIterator, Dict, List
+from enum import Enum
+from dataclasses import dataclass
+
+
+class SteeringPriority(Enum):
+    """Priority levels for steering messages."""
+    LOW = 1
+    NORMAL = 5
+    HIGH = 10
+    URGENT = 20
+    INTERRUPT = 30  # Immediate interruption
+
+
+@dataclass
+class SteeringMessage:
+    """A steering message with priority and metadata."""
+    content: str
+    priority: SteeringPriority = SteeringPriority.NORMAL
+    metadata: Optional[Dict[str, Any]] = None
 
 
 @runtime_checkable
@@ -431,6 +450,140 @@ class ManagedBackendProtocol(Protocol):
         ...
 
 
+@runtime_checkable
+class MessageSteeringProtocol(Protocol):
+    """
+    Protocol for message steering during agent execution.
+    
+    This protocol enables real-time communication with agents during
+    long-running operations, allowing users to provide mid-execution
+    guidance, feedback, and course corrections.
+    
+    Example:
+        ```python
+        class BasicMessageSteering:
+            def queue_message(self, message: str, priority: int = 5) -> str:
+                # Add to internal queue
+                return "msg_001"
+            
+            def get_pending_messages(self) -> List[SteeringMessage]:
+                # Return queued messages
+                return []
+            
+            def process_steering(self, context: dict) -> bool:
+                # Check for messages and update context
+                return False
+        
+        agent = Agent(name="assistant", message_steering=BasicMessageSteering())
+        ```
+    """
+    
+    def queue_message(self, message: str, priority: int = 5) -> str:
+        """
+        Queue a steering message for the agent.
+        
+        Args:
+            message: The steering message content
+            priority: Message priority (higher = more urgent)
+            
+        Returns:
+            Message ID for tracking
+        """
+        ...
+    
+    def get_pending_messages(self) -> List[Any]:
+        """
+        Get all pending steering messages.
+        
+        Returns:
+            List of pending messages ordered by priority
+        """
+        ...
+    
+    def process_steering(self, context: Optional[Dict[str, Any]] = None) -> bool:
+        """
+        Process any pending steering messages and update execution context.
+        
+        This method is called during agent execution to check for
+        and process any steering messages that have been queued.
+        
+        Args:
+            context: Optional execution context to update
+            
+        Returns:
+            True if messages were processed, False otherwise
+        """
+        ...
+    
+    def clear_messages(self) -> int:
+        """
+        Clear all pending messages.
+        
+        Returns:
+            Number of messages cleared
+        """
+        ...
+    
+    def has_pending_messages(self) -> bool:
+        """
+        Check if there are pending steering messages.
+        
+        Returns:
+            True if messages are pending, False otherwise
+        """
+        ...
+    
+    @property
+    def enabled(self) -> bool:
+        """
+        Whether message steering is enabled.
+        
+        Returns:
+            True if steering is enabled, False otherwise
+        """
+        ...
+
+
+@runtime_checkable
+class AgentSteeringProtocol(Protocol):
+    """
+    Protocol for agents that support message steering.
+    
+    This extends the agent interface with steering capabilities,
+    allowing real-time message injection during execution.
+    """
+    
+    def steer(self, message: str, priority: int = 5) -> str:
+        """
+        Send a steering message to the agent during execution.
+        
+        This can be called while the agent is running to provide
+        real-time guidance or course corrections.
+        
+        Args:
+            message: The steering message
+            priority: Message priority (higher = more urgent)
+            
+        Returns:
+            Message ID for tracking
+        """
+        ...
+    
+    def get_steering_status(self) -> Dict[str, Any]:
+        """
+        Get current steering status.
+        
+        Returns:
+            Dict with pending message count, last processed time, etc.
+        """
+        ...
+    
+    @property
+    def message_steering_enabled(self) -> bool:
+        """Whether message steering is enabled for this agent."""
+        ...
+
+
 __all__ = [
     'AgentProtocol',
     'RunnableAgentProtocol', 
@@ -441,4 +594,8 @@ __all__ = [
     'HttpLauncherProtocol',
     'McpLauncherProtocol',
     'ManagedBackendProtocol',
+    'SteeringPriority',
+    'SteeringMessage',
+    'MessageSteeringProtocol',
+    'AgentSteeringProtocol',
 ]

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

@@ -0,0 +1,231 @@
+"""
+Sandbox execution mixin for Agent.
+
+Provides safe code execution capabilities to agents via the sandbox framework.
+"""
+
+import logging
+from typing import Optional, Dict, Any, Union
+
+from ..sandbox import (
+    SandboxConfig, 
+    SandboxManager, 
+    SandboxResult,
+    check_code_safety,
+    format_warnings,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class SandboxMixin:
+    """Mixin that adds sandbox execution capabilities to Agent.
+    
+    Provides safe code execution in isolated environments with security pre-checks.
+    """
+    
+    def __init__(self, *args, **kwargs):
+        # Extract sandbox config from kwargs
+        self.sandbox_config: Optional[SandboxConfig] = kwargs.pop('sandbox', None)
+        
+        # Convert boolean True to default config
+        if self.sandbox_config is True:
+            self.sandbox_config = SandboxConfig.subprocess()
+        
+        super().__init__(*args, **kwargs)
+        
+        self._sandbox_manager: Optional[SandboxManager] = None
+    
+    @property
+    def has_sandbox(self) -> bool:
+        """Check if sandbox is configured for this agent."""
+        return self.sandbox_config is not None
+    
+    def get_sandbox_manager(self) -> Optional[SandboxManager]:
+        """Get or create sandbox manager."""
+        if not self.has_sandbox:
+            return None
+        
+        if self._sandbox_manager is None:
+            self._sandbox_manager = SandboxManager(self.sandbox_config)
+        
+        return self._sandbox_manager
+    
+    async def execute_code(
+        self,
+        code: str,
+        language: str = "python",
+        check_security: bool = True,
+        **kwargs
+    ) -> SandboxResult:
+        """Execute code safely in configured sandbox.
+        
+        Args:
+            code: Code to execute
+            language: Programming language (python, bash, etc.)
+            check_security: Whether to run security pre-checks
+            **kwargs: Additional arguments passed to sandbox.execute()
+            
+        Returns:
+            SandboxResult with execution details
+            
+        Raises:
+            RuntimeError: If no sandbox is configured
+        """
+        if not self.has_sandbox:
+            raise RuntimeError(
+                "No sandbox configured. Set sandbox=True or provide SandboxConfig to Agent()"
+            )
+        
+        # Security pre-check if enabled
+        if check_security:
+            warnings = check_code_safety(code, language)
+            if warnings:
+                warning_text = format_warnings(warnings)
+                if self.verbose:
+                    logger.warning(f"Security warnings for code execution:\n{warning_text}")
+                
+                # Store warnings in result metadata
+                kwargs.setdefault('metadata', {})['security_warnings'] = warnings
+        
+        manager = self.get_sandbox_manager()
+        return await manager.run_code(code, language=language, **kwargs)
+    
+    def execute_code_sync(
+        self,
+        code: str,
+        language: str = "python", 
+        check_security: bool = True,
+        **kwargs
+    ) -> SandboxResult:
+        """Synchronous wrapper for execute_code.
+        
+        Args:
+            code: Code to execute
+            language: Programming language
+            check_security: Whether to run security pre-checks
+            **kwargs: Additional arguments
+            
+        Returns:
+            SandboxResult with execution details
+        """
+        from ..approval.utils import run_coroutine_safely
+        return run_coroutine_safely(self.execute_code(code, language, check_security, **kwargs))
+    
+    async def run_shell_command(
+        self,
+        command: Union[str, list],
+        check_security: bool = True,
+        **kwargs
+    ) -> SandboxResult:
+        """Run a shell command in the sandbox.
+        
+        Args:
+            command: Shell command to run
+            check_security: Whether to run security pre-checks
+            **kwargs: Additional arguments
+            
+        Returns:
+            SandboxResult with execution details
+        """
+        if not self.has_sandbox:
+            raise RuntimeError("No sandbox configured")
+        
+        # Convert command to string for security checking
+        if isinstance(command, list):
+            command_str = " ".join(command)
+        else:
+            command_str = command
+        
+        # Security pre-check
+        if check_security:
+            warnings = check_code_safety(command_str, "bash")
+            if warnings:
+                warning_text = format_warnings(warnings)
+                if self.verbose:
+                    logger.warning(f"Security warnings for command:\n{warning_text}")
+        
+        async with self.get_sandbox_manager() as sandbox:
+            return await sandbox.run_command(command, **kwargs)
+    
+    def get_sandbox_status(self) -> Dict[str, Any]:
+        """Get sandbox status information.
+        
+        Returns:
+            Dictionary with sandbox status details
+        """
+        if not self.has_sandbox:
+            return {"configured": False}
+        
+        status = {
+            "configured": True,
+            "config": self.sandbox_config.to_dict() if self.sandbox_config else None,
+        }
+        
+        manager = self.get_sandbox_manager()
+        if manager:
+            # Get available sandbox types
+            available_types = manager.get_available_types()
+            status["available_types"] = available_types
+            status["current_type"] = self.sandbox_config.sandbox_type
+        
+        return status
+    
+    async def sandbox_cleanup(self):
+        """Clean up sandbox resources."""
+        if self._sandbox_manager:
+            # The context manager handles cleanup, but we can force it here
+            try:
+                await self._sandbox_manager.__aexit__(None, None, None)
+            except Exception as e:
+                logger.warning(f"Error during sandbox cleanup: {e}")
+            finally:
+                self._sandbox_manager = None
+    
+    def _get_code_execution_tools(self):
+        """Get sandbox-based code execution tools for this agent.
+        
+        Returns tools that use the configured sandbox for safe execution.
+        
+        Returns:
+            List of tool functions
+        """
+        if not self.has_sandbox:
+            return []
+        
+        from ..tools import tool
+        
+        @tool
+        def execute_python_code(code: str) -> str:
+            """Execute Python code safely in sandbox.
+            
+            Args:
+                code: Python code to execute
+                
+            Returns:
+                Execution output
+            """
+            result = self.execute_code_sync(code, language="python")
+            if result.success:
+                return result.stdout or result.output
+            else:
+                return f"Error: {result.error or result.stderr}"
+        
+        @tool  
+        def execute_shell_command(command: str) -> str:
+            """Execute shell command safely in sandbox.
+            
+            Args:
+                command: Shell command to execute
+                
+            Returns:
+                Command output
+            """
+            from ..approval.utils import run_coroutine_safely
+            result = run_coroutine_safely(self.run_shell_command(command))
+            if result.success:
+                return result.stdout or result.output
+            else:
+                return f"Error: {result.error or result.stderr}"
+        
+        return [execute_python_code, execute_shell_command]

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

@@ -16,6 +16,7 @@ import contextvars
 import concurrent.futures
 from typing import List, Optional, Any, Dict, Union, TYPE_CHECKING
 from ..errors import ToolExecutionError
+from ..tools.trust import wrap_if_external
 
 logger = logging.getLogger(__name__)
 
@@ -26,6 +27,21 @@ if TYPE_CHECKING:
 class ToolExecutionMixin:
     """Mixin providing toolexecution methods for the Agent class."""
 
+    def _get_existing_stream_emitter(self):
+        """Return an already-initialized stream emitter without creating one."""
+        emitter = getattr(self, "_stream_emitter", None)
+        if emitter is not None:
+            return emitter
+
+        # Support name-mangled private attributes across class renames/inheritance.
+        for cls in type(self).mro():
+            mangled = f"_{cls.__name__}__stream_emitter"
+            if hasattr(self, mangled):
+                emitter = getattr(self, mangled, None)
+                if emitter is not None:
+                    return emitter
+        return None
+
     def _resolve_tool_names(self, tool_names):
         """Resolve tool names to actual tool instances from registry.
         
@@ -120,6 +136,12 @@ class ToolExecutionMixin:
         """
         logging.debug(f"{self.name} executing tool {function_name} with arguments: {arguments}")
         
+        # Handle bridge tool unwrapping BEFORE trace/stream/hooks (design invariant #6)
+        # Only intercept when tool_search is active; otherwise fall through to real tool execution
+        if (getattr(self, '_tool_search_config', None) is not None and
+                function_name in ("tool_search", "tool_describe", "tool_call")):
+            return self._handle_bridge_tool_call(function_name, arguments, tool_call_id)
+        
         # NOTE: tool_call callback is triggered by display_tool_call in openai_client.py
         # Do NOT call it here to avoid duplicate output
         
@@ -160,8 +182,9 @@ class ToolExecutionMixin:
         
         # Emit TOOL_CALL_START to stream_emitter (for AIUI/AG-UI consumers)
         # Zero overhead when no callbacks registered
-        if hasattr(self, '_Agent__stream_emitter') and getattr(self, "_Agent__stream_emitter", None) is not None and getattr(self, "_Agent__stream_emitter", None).has_callbacks:
-            getattr(self, "_Agent__stream_emitter", None).emit(StreamEvent(
+        _stream_emitter = self._get_existing_stream_emitter()
+        if _stream_emitter is not None and _stream_emitter.has_callbacks:
+            _stream_emitter.emit(StreamEvent(
                 type=StreamEventType.TOOL_CALL_START,
                 timestamp=_tool_start_perf,
                 tool_call={
@@ -173,6 +196,19 @@ class ToolExecutionMixin:
             ))
         
         try:
+            # Check for steering messages before tool execution
+            if hasattr(self, '_check_steering_messages'):
+                try:
+                    steering_msg = self._check_steering_messages()
+                    if steering_msg:
+                        # Check if steering message indicates interruption priority
+                        # (SteeringMixin formats HIGH/URGENT as "[URGENT USER GUIDANCE]" and INTERRUPT as "[INTERRUPT USER GUIDANCE]")
+                        if "[URGENT USER GUIDANCE]" in steering_msg or "[INTERRUPT USER GUIDANCE]" in steering_msg:
+                            logger.info(f"Tool {function_name} execution interrupted by high-priority steering")
+                            return f"Tool execution interrupted by user guidance: {steering_msg}"
+                except Exception as e:
+                    logger.warning(f"Steering check failed, continuing with tool execution: {e}")
+            
             # Trigger BEFORE_TOOL hook
             from ..hooks import HookEvent, BeforeToolInput
             before_tool_input = BeforeToolInput(
@@ -194,6 +230,29 @@ class ToolExecutionMixin:
                 if res.output and res.output.modified_data:
                     arguments.update(res.output.modified_data)
 
+            # Loop guard check - prevent tool execution loops with graduated response
+            if hasattr(self, '_loop_guard') and self._loop_guard:
+                from ..escalation.loop_guard import GuardAction
+                decision = self._loop_guard.check(function_name, arguments, is_pre_execution=True)
+                
+                if decision.action == GuardAction.WARN:
+                    # Inject warning into tool result so LLM sees guidance
+                    logging.warning(f"Loop guard warning for {function_name}: {decision.message}")
+                elif decision.action == GuardAction.BLOCK:
+                    # Block tool execution but continue through teardown path
+                    logging.warning(f"Loop guard blocked {function_name}: {decision.message}")
+                    # Set flag to use blocked result instead of executing tool
+                    blocked_result = {"error": f"[loop-guard] {decision.message}", "loop_blocked": True}
+                elif decision.action == GuardAction.HALT:
+                    # Halt execution with exception
+                    from ..errors import ToolExecutionError
+                    raise ToolExecutionError(
+                        f"[loop-guard] {decision.message}",
+                        tool_name=function_name,
+                        agent_id=self.name,
+                        is_retryable=False
+                    )
+
             # C4 — optional tool-argument validation via ToolValidatorProtocol.
             # Zero overhead when not set. Users wire via `agent._tool_validator = MyValidator()`.
             _validator = getattr(self, '_tool_validator', None)
@@ -209,32 +268,41 @@ class ToolExecutionMixin:
                 except Exception as _ve:  # noqa: BLE001 — never break tool exec on validator bug
                     logging.debug(f"Tool validator raised; skipping validation: {_ve}")
 
-            # P8/G11: Apply tool timeout if configured
-            tool_timeout = getattr(self, '_tool_timeout', None)
-            if tool_timeout and tool_timeout > 0:
-                # Use copy_context to preserve injection context in executor thread
-                ctx = contextvars.copy_context()
-                
-                def execute_with_context():
-                    with with_injection_context(state):
-                        return self._execute_tool_with_circuit_breaker(function_name, arguments)
-                
-                # Use reusable executor to prevent resource leaks
-                if not hasattr(self, '_tool_executor'):
-                    self._tool_executor = concurrent.futures.ThreadPoolExecutor(
-                        max_workers=2, thread_name_prefix=f"tool-{self.name}"
-                    )
-                
-                future = self._tool_executor.submit(ctx.run, execute_with_context)
-                try:
-                    result = future.result(timeout=tool_timeout)
-                except concurrent.futures.TimeoutError:
-                    future.cancel()
-                    logging.warning(f"Tool {function_name} timed out after {tool_timeout}s")
-                    result = {"error": f"Tool timed out after {tool_timeout}s", "timeout": True}
+            # Check if loop guard blocked execution
+            blocked_result = locals().get('blocked_result')
+            if blocked_result is not None:
+                result = blocked_result
             else:
-                with with_injection_context(state):
-                    result = self._execute_tool_with_circuit_breaker(function_name, arguments)
+                # P8/G11: Apply tool timeout if configured
+                tool_timeout = getattr(self, '_tool_timeout', None)
+                if tool_timeout and tool_timeout > 0:
+                    # Use copy_context to preserve injection context in executor thread
+                    ctx = contextvars.copy_context()
+                    
+                    def execute_with_context():
+                        with with_injection_context(state):
+                            return self._execute_tool_with_circuit_breaker(function_name, arguments)
+                    
+                    # Use reusable executor to prevent resource leaks
+                    if not hasattr(self, '_tool_executor'):
+                        self._tool_executor = concurrent.futures.ThreadPoolExecutor(
+                            max_workers=2, thread_name_prefix=f"tool-{self.name}"
+                        )
+                    
+                    future = self._tool_executor.submit(ctx.run, execute_with_context)
+                    try:
+                        result = future.result(timeout=tool_timeout)
+                    except concurrent.futures.TimeoutError:
+                        future.cancel()
+                        logging.warning(f"Tool {function_name} timed out after {tool_timeout}s")
+                        result = {"error": f"Tool timed out after {tool_timeout}s", "timeout": True}
+                else:
+                    with with_injection_context(state):
+                        result = self._execute_tool_with_circuit_breaker(function_name, arguments)
+            
+            # Apply prompt injection protection for external tools
+            # Zero-cost for trusted tools, wraps external content in security markers
+            result = wrap_if_external(function_name, result)
             
             # Apply tool output truncation to prevent context overflow
             # Uses context manager budget if enabled, otherwise applies default limit
@@ -275,10 +343,10 @@ class ToolExecutionMixin:
             
             # Emit TOOL_CALL_RESULT to stream_emitter (for AIUI/AG-UI consumers)
             # Zero overhead when no callbacks registered
-            if hasattr(self, '_Agent__stream_emitter') and getattr(self, "_Agent__stream_emitter", None) is not None and getattr(self, "_Agent__stream_emitter", None).has_callbacks:
+            if _stream_emitter is not None and _stream_emitter.has_callbacks:
                 # Truncate result for stream event (keep it reasonable for UI display)
                 result_summary = str(result)[:500] if result else None
-                getattr(self, "_Agent__stream_emitter", None).emit(StreamEvent(
+                _stream_emitter.emit(StreamEvent(
                     type=StreamEventType.TOOL_CALL_RESULT,
                     timestamp=_time.perf_counter(),
                     tool_call={
@@ -290,6 +358,18 @@ class ToolExecutionMixin:
                     agent_id=self.name,
                     metadata={"duration_ms": _duration_ms},
                 ))
+                _stream_emitter.emit(StreamEvent(
+                    type=StreamEventType.TOOL_CALL_END,
+                    timestamp=_time.perf_counter(),
+                    tool_call={
+                        "name": function_name,
+                        "arguments": arguments,
+                        "result": result_summary,
+                        "id": tool_call_id,
+                    },
+                    agent_id=self.name,
+                    metadata={"duration_ms": _duration_ms},
+                ))
             
             # Trigger AFTER_TOOL hook
             from ..hooks import HookEvent, AfterToolInput
@@ -313,6 +393,25 @@ class ToolExecutionMixin:
                 if not is_error:
                     self._doom_loop_tracker.mark_progress(f"tool:{function_name}")
             
+            # Record tool execution in loop guard
+            if hasattr(self, '_loop_guard') and self._loop_guard:
+                is_success = result is not None and not (isinstance(result, dict) and result.get('error'))
+                self._loop_guard.record(function_name, arguments, is_success)
+                # Handle warning injection for WARN decisions
+                decision = self._loop_guard.check(function_name, arguments, is_pre_execution=False) 
+                if decision.action.value == "warn":
+                    if isinstance(result, str):
+                        result = f"{result}\n\n[loop-guard] {decision.message}"
+                    elif isinstance(result, dict):
+                        # Inject warning into dict results
+                        result["_loop_guard"] = {"message": decision.message, "action": decision.action.value}
+                    elif isinstance(result, list):
+                        # Inject warning into list results  
+                        result = {"value": result, "_loop_guard": {"message": decision.message, "action": decision.action.value}}
+                    else:
+                        # Wrap non-string/dict/list results to preserve original data plus warning
+                        result = {"value": result, "_loop_guard": {"message": decision.message, "action": decision.action.value}}
+            
             # Increment per-turn tool count for no-tool-call detection
             self._autonomy_turn_tool_count = getattr(self, '_autonomy_turn_tool_count', 0) + 1
             
@@ -322,6 +421,10 @@ class ToolExecutionMixin:
             _duration_ms = (_time.time() - _tool_start_time) * 1000
             _trace_emitter.tool_call_end(self.name, function_name, None, _duration_ms, str(e))
             
+            # Preserve existing ToolExecutionError unchanged to maintain loop guard HALT behavior
+            if isinstance(e, ToolExecutionError):
+                raise
+            
             # Gap 3a fix: Wrap exceptions in ToolExecutionError for better observability
             is_retryable = not isinstance(e, (ValueError, TypeError, AttributeError))
             raise ToolExecutionError(
@@ -610,7 +713,111 @@ class ToolExecutionMixin:
         return None, arguments
 
     def _execute_tool_with_circuit_breaker(self, function_name, arguments):
-        """Execute tool with circuit breaker protection.
+        """Execute tool with retry policy and circuit breaker protection.
+        
+        Args:
+            function_name: Name of the tool to execute
+            arguments: Arguments for the tool
+            
+        Returns:
+            Tool execution result or circuit breaker error
+        """
+        # Get retry policy (tool-level > agent-level > default)
+        retry_policy = self._get_tool_retry_policy(function_name)
+        
+        last_exception = None
+        
+        # Retry loop with exponential backoff
+        for attempt in range(retry_policy.max_attempts):
+            try:
+                result = self._execute_tool_with_circuit_breaker_impl(function_name, arguments)
+                
+                # Check if result is an error that should be retried
+                if isinstance(result, dict) and result.get("error"):
+                    # Skip retry for non-retryable errors (approval, permission, etc.)
+                    if (result.get("approval_denied") or 
+                        result.get("permission_denied") or 
+                        result.get("approval_error") or
+                        result.get("circuit_open")):
+                        return result
+                    
+                    # Determine error type for retry policy
+                    error_type = self._classify_error_type(result, last_exception)
+                    
+                    # Check if we should retry
+                    if not retry_policy.should_retry(error_type, attempt):
+                        return result
+                    
+                    # Don't retry on last attempt
+                    if attempt == retry_policy.max_attempts - 1:
+                        return result
+                    
+                    # Emit retry hook event  
+                    delay_ms = retry_policy.get_delay_ms(attempt)
+                    self._emit_retry_hook(function_name, attempt + 1, delay_ms, result.get("error", "Unknown error"), retry_policy.max_attempts, error_type)
+                    
+                    # Wait before retry
+                    time.sleep(delay_ms / 1000.0)
+                    continue
+                else:
+                    # Success - return result
+                    return result
+                    
+            except ToolExecutionError as e:
+                last_exception = e
+                # Check if the error is retryable
+                if not e.is_retryable or attempt == retry_policy.max_attempts - 1:
+                    raise
+                
+                # Determine error type
+                error_type = self._classify_error_type(None, e) 
+                
+                if not retry_policy.should_retry(error_type, attempt):
+                    raise
+                
+                # Emit retry hook event
+                delay_ms = retry_policy.get_delay_ms(attempt)
+                self._emit_retry_hook(function_name, attempt + 1, delay_ms, str(e), retry_policy.max_attempts, error_type)
+                
+                # Wait before retry
+                time.sleep(delay_ms / 1000.0)
+                continue
+                
+            except Exception as e:
+                # Wrap in ToolExecutionError for consistency
+                is_retryable = not isinstance(e, (ValueError, TypeError, AttributeError))
+                wrapped_error = ToolExecutionError(
+                    f"Tool '{function_name}' failed: {e}",
+                    tool_name=function_name,
+                    agent_id=self.name,
+                    is_retryable=is_retryable,
+                )
+                
+                # Check if retryable and not last attempt
+                if not is_retryable or attempt == retry_policy.max_attempts - 1:
+                    raise wrapped_error from e
+                
+                # Determine error type  
+                error_type = self._classify_error_type(None, wrapped_error)
+                
+                if not retry_policy.should_retry(error_type, attempt):
+                    raise wrapped_error from e
+                
+                # Emit retry hook event
+                delay_ms = retry_policy.get_delay_ms(attempt)
+                self._emit_retry_hook(function_name, attempt + 1, delay_ms, str(e), retry_policy.max_attempts, error_type)
+                
+                # Wait before retry
+                time.sleep(delay_ms / 1000.0)
+                continue
+        
+        # Should never reach here due to loop logic, but safety fallback
+        if last_exception:
+            raise last_exception
+        return {"error": "Maximum retry attempts exceeded"}
+
+    def _execute_tool_with_circuit_breaker_impl(self, function_name, arguments):
+        """Execute tool with circuit breaker protection (internal implementation).
         
         Args:
             function_name: Name of the tool to execute
@@ -926,4 +1133,204 @@ class ToolExecutionMixin:
     def pending_approval_count(self) -> int:
         """Number of approval requests still waiting."""
         return len(self._pending_approvals)
+    
+    def _handle_bridge_tool_call(self, function_name: str, arguments: Dict[str, Any], tool_call_id: Optional[str] = None) -> Any:
+        """
+        Handle bridge tool calls (tool_search, tool_describe, tool_call).
+        
+        This implements the tool search unwrapping logic before trace/stream/hooks
+        as required by design invariant #6.
+        
+        Args:
+            function_name: Bridge tool name
+            arguments: Arguments passed to bridge tool
+            tool_call_id: Optional tool call ID
+            
+        Returns:
+            Result of bridge tool execution or unwrapped real tool call
+        """
+        # Ensure tool search metadata is available
+        if not hasattr(self, '_tool_search_metadata') or self._tool_search_metadata is None:
+            return "Tool search not available or not in bridge mode"
+        
+        metadata = self._tool_search_metadata
+        
+        # Check if we're in bridge mode
+        if not metadata.get("bridge_mode", False):
+            return "Tool search not in bridge mode"
+        
+        # Get deferrable tools from metadata
+        deferrable_tools = metadata.get("deferrable_tools", [])
+        
+        if function_name == "tool_search":
+            # Handle tool_search bridge call
+            try:
+                from ..tools.tool_search import dispatch_tool_search
+                query = arguments.get("query", "")
+                limit = arguments.get("limit", None)
+                
+                result = dispatch_tool_search(
+                    query=query,
+                    limit=limit, 
+                    deferrable_tools=deferrable_tools,
+                    config=self._tool_search_config
+                )
+                return json.dumps(result, indent=2)
+            except ImportError:
+                return "Tool search module not available"
+            except Exception as e:
+                logging.error(f"Error in tool_search: {e}")
+                return f"Error searching tools: {e}"
+        
+        elif function_name == "tool_describe":
+            # Handle tool_describe bridge call
+            try:
+                from ..tools.tool_search import dispatch_tool_describe
+                tool_name = arguments.get("tool_name", "")
+                
+                result = dispatch_tool_describe(
+                    tool_name=tool_name,
+                    deferrable_tools=deferrable_tools
+                )
+                return json.dumps(result, indent=2)
+            except ImportError:
+                return "Tool search module not available"
+            except Exception as e:
+                logging.error(f"Error in tool_describe: {e}")
+                return f"Error describing tool: {e}"
+        
+        elif function_name == "tool_call":
+            # Handle tool_call bridge - unwrap and recurse with real tool
+            try:
+                from ..tools.tool_search import resolve_underlying_call
+                
+                # Unwrap the real tool call
+                real_function_name, real_arguments = resolve_underlying_call(function_name, arguments)
+                
+                # Validate that the real tool is in our deferrable set (security check)
+                deferrable_names = {
+                    tool_def.get("function", {}).get("name", "") 
+                    for tool_def in deferrable_tools
+                }
+                
+                if real_function_name not in deferrable_names:
+                    return f"Tool '{real_function_name}' is not available for execution"
+                
+                # Recursively execute the real tool (this will go through normal execution path)
+                return self.execute_tool(real_function_name, real_arguments, tool_call_id)
+                
+            except ImportError:
+                return "Tool search module not available"
+            except Exception as e:
+                logging.error(f"Error in tool_call unwrap: {e}")
+                return f"Error executing tool: {e}"
+        
+        else:
+            return f"Unknown bridge tool: {function_name}"
 
+    def _get_tool_retry_policy(self, tool_name):
+        """Get retry policy for a tool (tool-level > agent-level > default).
+        
+        Args:
+            tool_name: Name of the tool to get retry policy for
+            
+        Returns:
+            RetryPolicy instance
+        """
+        from ..tools.retry import RetryPolicy
+        
+        # Check for tool-level retry policy first
+        tools = getattr(self, 'tools', [])
+        # Handle non-iterable tools (e.g., single MCP instance)
+        if not isinstance(tools, (list, tuple)):
+            tools = []  # MCP or single tool instance - no tool-level policy lookup
+        for tool in tools:
+            if (callable(tool) and 
+                getattr(tool, '__name__', '') == tool_name and
+                hasattr(tool, 'retry_policy')):
+                return tool.retry_policy
+        
+        # Check for agent-level retry policy
+        agent_policy = getattr(self, '_tool_retry_policy', None)
+        if agent_policy is not None:
+            return agent_policy
+        
+        # Return default retry policy (cached class-level instance)
+        if not hasattr(ToolExecutionMixin, '_default_retry_policy'):
+            ToolExecutionMixin._default_retry_policy = RetryPolicy()
+        return ToolExecutionMixin._default_retry_policy
+
+    def _classify_error_type(self, error_dict, exception):
+        """Classify error type for retry policy matching.
+        
+        Args:
+            error_dict: Error dictionary from tool execution (if any)
+            exception: Exception that was raised (if any)
+            
+        Returns:
+            String error type for retry policy checking
+        """
+        # Check error dict first
+        if error_dict and isinstance(error_dict, dict):
+            error_msg = error_dict.get("error", "").lower()
+            if "timeout" in error_msg or "timed out" in error_msg:
+                return "timeout"
+            elif "rate" in error_msg and "limit" in error_msg:
+                return "rate_limit"
+            elif "connection" in error_msg or "network" in error_msg:
+                return "connection_error"
+        
+        # Check exception type
+        if exception:
+            exc_msg = str(exception).lower()
+            exc_type = type(exception).__name__.lower()
+            
+            if "timeout" in exc_msg or "timeout" in exc_type:
+                return "timeout"
+            elif "rate" in exc_msg and "limit" in exc_msg:
+                return "rate_limit"  
+            elif ("connection" in exc_msg or "network" in exc_msg or 
+                  "connection" in exc_type):
+                return "connection_error"
+        
+        return "unknown"
+
+    def _emit_retry_hook(self, tool_name, attempt, delay_ms, error, max_attempts, error_type):
+        """Emit ON_RETRY hook event.
+        
+        Args:
+            tool_name: Name of the tool being retried
+            attempt: Current attempt number (1-based)
+            delay_ms: Delay before retry in milliseconds
+            error: Error message or description
+            max_attempts: Maximum number of attempts configured
+            error_type: Classified error type
+        """
+        try:
+            from ..hooks import HookEvent, OnRetryInput
+            
+            # Only emit if we have a hook runner
+            hook_runner = getattr(self, '_hook_runner', None)
+            if hook_runner is None:
+                return
+            
+            retry_input = OnRetryInput(
+                session_id=getattr(self, '_session_id', 'default'),
+                cwd=os.getcwd(),
+                event_name=HookEvent.ON_RETRY,
+                timestamp=str(time.time()),
+                agent_name=self.name,
+                tool_name=tool_name,
+                attempt=attempt,
+                delay_ms=delay_ms,
+                error=error,
+                max_attempts=max_attempts,
+                error_type=error_type
+            )
+            
+            # Execute hook synchronously
+            hook_runner.execute_sync(HookEvent.ON_RETRY, retry_input, target=tool_name)
+            
+        except Exception as e:
+            # Don't let hook failures break retry logic
+            logging.debug(f"Failed to emit retry hook: {e}")

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

@@ -77,6 +77,10 @@ class UnifiedExecutionMixin:
         _trace_emitter = get_context_emitter()
         _trace_emitter.agent_start(self.name, {"role": self.role, "goal": self.goal})
         
+        # Reset loop guard for new chat turn (always-on protection)
+        if hasattr(self, '_loop_guard') and self._loop_guard:
+            self._loop_guard.reset_turn()
+        
         try:
             # CLI Backend routing - delegate entire turn if configured
             if hasattr(self, '_cli_backend') and self._cli_backend is not None:

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

@@ -8,7 +8,8 @@ from .autoagents import AutoAgents
 from .auto_rag_agent import AutoRagAgent, AutoRagConfig, RetrievalPolicy
 from .protocols import (
     MergeStrategyProtocol, FirstWinsMerge, ConcatMerge, DictMerge,
-    TaskExecutorProtocol, ExecutionContext, TaskResult
+    TaskExecutorProtocol, ExecutionContext, TaskResult,
+    SpawnAnnounceProtocol, SpawnedSubAgent, SubAgentCompletionEvent
 )
 
 __all__ = [
@@ -20,4 +21,5 @@ __all__ = [
     'AutoRagAgent', 'AutoRagConfig', 'RetrievalPolicy',
     'MergeStrategyProtocol', 'FirstWinsMerge', 'ConcatMerge', 'DictMerge',
     'TaskExecutorProtocol', 'ExecutionContext', 'TaskResult',
+    'SpawnAnnounceProtocol', 'SpawnedSubAgent', 'SubAgentCompletionEvent',
 ]

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

@@ -4,20 +4,23 @@ import json
 import logging
 import threading
 from praisonaiagents._logging import get_logger
-from typing import Any, Dict, Optional, List, Tuple
+from typing import Any, Dict, Optional, List, Tuple, Callable
 from ..main import display_error, TaskOutput
 from ..agent.agent import Agent
 from ..task.task import Task
 from ..process.process import Process
+from .protocols import SpawnAnnounceProtocol, SpawnedSubAgent, SubAgentCompletionEvent
+from ..bus.bus import EventBus
+from ..bus.event import EventType, Event
 import asyncio
 import uuid
 from enum import Enum
 
 # Import token tracking
 try:
-    from ..telemetry.token_collector import _token_collector
+    from ..telemetry.token_collector import get_token_collector
 except ImportError:
-    _token_collector = None
+    get_token_collector = None
 
 # Import async utility for hot-path usage
 try:
@@ -126,7 +129,24 @@ class _AgentServerRegistry:
         
         thread = threading.Thread(target=run_server, daemon=True)
         thread.start()
-        ready_event.wait(timeout=5.0)  # Deterministic wait instead of sleep(0.5)
+        
+        # Check for configurable timeout via environment variable
+        try:
+            timeout = float(os.environ.get("PRAISONAI_SERVER_READY_TIMEOUT", "5.0"))
+        except ValueError:
+            logger.warning("Invalid PRAISONAI_SERVER_READY_TIMEOUT value. Using default 5.0s.")
+            timeout = 5.0
+        became_ready = ready_event.wait(timeout=timeout)
+        
+        if not became_ready:
+            logger.warning(
+                "Agent server on port %s did not become ready within %.1fs. "
+                "Proceeding, but some features may not work correctly. "
+                "Check server logs for startup errors.",
+                port,
+                timeout,
+            )
+        
         return True
 
 
@@ -503,7 +523,7 @@ def process_task_context(context_item, verbose=0, user_id=None):
     else:
         return str(context_item)  # Fallback for unknown types
 
-class AgentTeam:
+class AgentTeam(SpawnAnnounceProtocol):
     """
     Multi-agent coordinator that manages and delegates work to multiple agents.
     
@@ -778,6 +798,14 @@ class AgentTeam:
         self.on_task_complete = _on_task_complete
         self.variables = variables if variables else {}
         
+        # Spawn-announce pattern support
+        self._spawned_agents: Dict[str, SpawnedSubAgent] = {}
+        self._completion_callbacks: Dict[str, Callable[[SubAgentCompletionEvent], Any]] = {}
+        self._completion_events: List[SubAgentCompletionEvent] = []
+        self._event_bus: Optional[EventBus] = None
+        self._spawn_lock = threading.RLock()  # Thread-safe spawn operations (reentrant)
+        self._team_id = str(uuid.uuid4())  # Unique team identifier
+        
         # Check for manager_llm in environment variable if not provided
         self.manager_llm = manager_llm or os.getenv('OPENAI_MODEL_NAME', 'gpt-4o-mini')
         
@@ -1592,25 +1620,25 @@ class AgentTeam:
             original_on_task_start = self.on_task_start
             original_on_task_complete = self.on_task_complete
             
-            def composed_on_task_start(task):
+            def composed_on_task_start(task, task_id):
                 try:
-                    verbose_task_start_callback(task)
+                    verbose_task_start_callback(task, task_id)
                 except Exception as e:
                     logging.debug(f"Error in verbose task start callback: {e}")
                 if original_on_task_start:
                     try:
-                        original_on_task_start(task)
+                        original_on_task_start(task, task_id)
                     except Exception as e:
                         logging.debug(f"Error in original task start callback: {e}")
             
-            def composed_on_task_complete(task):
+            def composed_on_task_complete(task, task_output):
                 try:
-                    verbose_task_complete_callback(task)
+                    verbose_task_complete_callback(task, task_output)
                 except Exception as e:
                     logging.debug(f"Error in verbose task complete callback: {e}")
                 if original_on_task_complete:
                     try:
-                        original_on_task_complete(task)
+                        original_on_task_complete(task, task_output)
                     except Exception as e:
                         logging.debug(f"Error in original task complete callback: {e}")
             
@@ -1809,18 +1837,19 @@ class AgentTeam:
 
     def get_token_usage_summary(self) -> Dict[str, Any]:
         """Get a summary of token usage across all agents and tasks."""
-        if not _token_collector:
+        if not get_token_collector:
             return {"error": "Token tracking not available"}
         
-        return _token_collector.get_session_summary()
+        return get_token_collector().get_session_summary()
     
     def get_detailed_token_report(self) -> Dict[str, Any]:
         """Get a detailed token usage report."""
-        if not _token_collector:
+        if not get_token_collector:
             return {"error": "Token tracking not available"}
         
-        summary = _token_collector.get_session_summary()
-        recent = _token_collector.get_recent_interactions(limit=20)
+        collector = get_token_collector()
+        summary = collector.get_session_summary()
+        recent = collector.get_recent_interactions(limit=20)
         
         # Calculate cost estimates (example rates)
         cost_per_1k_input = 0.0005  # $0.0005 per 1K input tokens
@@ -1844,11 +1873,11 @@ class AgentTeam:
     
     def display_token_usage(self):
         """Display token usage in a formatted table."""
-        if not _token_collector:
+        if not get_token_collector:
             print("Token tracking not available")
             return
         
-        summary = _token_collector.get_session_summary()
+        summary = get_token_collector().get_session_summary()
         
         print("\n" + "="*50)
         print("TOKEN USAGE SUMMARY")
@@ -2718,6 +2747,418 @@ class AgentTeam:
             except Exception as e:
                 logger.warning(f"Context manager cleanup failed: {e}")
 
+    # Spawn-Announce Protocol Implementation
+    def spawn_sub_agent(
+        self,
+        agent: Agent,
+        task: Any,
+        completion_callback: Optional[Callable[[SubAgentCompletionEvent], Any]] = None,
+        metadata: Optional[Dict[str, Any]] = None
+    ) -> SpawnedSubAgent:
+        """Spawn a sub-agent for non-blocking execution (sync version).
+        
+        Implements the SpawnAnnounceProtocol for efficient parallel sub-agent workflows.
+        Integrates with existing handoff infrastructure to avoid code duplication.
+        Uses threading.Thread for background execution.
+        
+        Note: For async contexts, use aspawn_sub_agent() which uses asyncio primitives.
+        This method is safe to call from sync contexts only.
+        """
+        with self._spawn_lock:
+            # Create unique IDs
+            agent_id = f"{self._team_id}_{str(uuid.uuid4())[:8]}"
+            task_id = f"task_{str(uuid.uuid4())[:8]}"
+            
+            # Ensure event bus is initialized
+            if self._event_bus is None:
+                self._event_bus = EventBus()
+                # Subscribe to completion events for this team
+                self._event_bus.subscribe(
+                    self._handle_sub_agent_completion, 
+                    [EventType.SUBAGENT_COMPLETED.value, EventType.SUBAGENT_ERROR.value]
+                )
+            
+            # Create spawned agent record
+            spawned = SpawnedSubAgent(
+                agent_id=agent_id,
+                task_id=task_id,
+                agent=agent,
+                task=task,
+                spawn_time=time.time(),
+                parent_id=self._team_id,
+                metadata=metadata or {}
+            )
+            
+            # Store spawn record and callback
+            self._spawned_agents[agent_id] = spawned
+            if completion_callback:
+                self._completion_callbacks[agent_id] = completion_callback
+            
+            # Publish spawn event
+            self._event_bus.publish(
+                EventType.SUBAGENT_SPAWNED.value,
+                {
+                    "agent_id": agent_id,
+                    "task_id": task_id,
+                    "parent_id": self._team_id,
+                    "agent_name": getattr(agent, 'name', 'unknown'),
+                    "spawn_time": spawned.spawn_time,
+                    "metadata": spawned.metadata
+                }
+            )
+            
+            # Start the sub-agent asynchronously
+            import threading
+            def _execute_sub_agent():
+                try:
+                    # Execute the task with the sub-agent
+                    if hasattr(task, 'description'):
+                        result = agent.start(task.description)
+                    else:
+                        result = agent.start(str(task))
+                    
+                    # Announce successful completion
+                    self.announce_completion(agent_id, task_id, result, success=True)
+                except Exception as e:
+                    # Announce failure
+                    self.announce_completion(agent_id, task_id, None, success=False, error=str(e))
+            
+            spawn_thread = threading.Thread(target=_execute_sub_agent, daemon=True)
+            spawn_thread.start()
+            
+            logger.debug(f"Spawned sub-agent {agent_id} for task {task_id}")
+            return spawned
+    
+    def announce_completion(
+        self,
+        agent_id: str,
+        task_id: str,
+        result: Any,
+        success: bool = True,
+        error: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None
+    ) -> None:
+        """Announce sub-agent completion via event bus (sync version).
+        
+        Note: For async contexts, use aannounce_completion() to avoid blocking the event loop.
+        """
+        with self._spawn_lock:
+            # Create completion event
+            completion_event = SubAgentCompletionEvent(
+                agent_id=agent_id,
+                task_id=task_id,
+                result=result,
+                success=success,
+                error=error,
+                parent_id=self._team_id,
+                metadata=metadata or {}
+            )
+            
+            # Store the completion event
+            self._completion_events.append(completion_event)
+            
+            # Publish completion event via event bus
+            if self._event_bus:
+                event_type = EventType.SUBAGENT_COMPLETED.value if success else EventType.SUBAGENT_ERROR.value
+                self._event_bus.publish(
+                    event_type,
+                    {
+                        "agent_id": agent_id,
+                        "task_id": task_id,
+                        "parent_id": self._team_id,
+                        "result": result,
+                        "success": success,
+                        "error": error,
+                        "completion_time": completion_event.completion_time,
+                        "metadata": completion_event.metadata
+                    }
+                )
+            
+            logger.debug(f"Announced completion for sub-agent {agent_id}: {'success' if success else 'error'}")
+    
+    def get_spawned_agents(self) -> List[SpawnedSubAgent]:
+        """Get list of currently spawned sub-agents."""
+        with self._spawn_lock:
+            return list(self._spawned_agents.values())
+    
+    def wait_for_completions(
+        self,
+        timeout: Optional[float] = None,
+        agent_ids: Optional[List[str]] = None
+    ) -> List[SubAgentCompletionEvent]:
+        """Wait for sub-agent completions (optional blocking method)."""
+        import time as time_module
+        start_time = time_module.time()
+        
+        # Read target agents inside lock to avoid race condition
+        with self._spawn_lock:
+            target_agents = agent_ids or list(self._spawned_agents.keys())
+        
+        completed_agents = set()
+        
+        while True:
+            # Check for completed agents
+            with self._spawn_lock:
+                for event in self._completion_events:
+                    if event.agent_id in target_agents:
+                        completed_agents.add(event.agent_id)
+                
+                # Return if all target agents are completed
+                if completed_agents >= set(target_agents):
+                    return [e for e in self._completion_events if e.agent_id in target_agents]
+            
+            # Check timeout
+            if timeout and (time_module.time() - start_time) > timeout:
+                break
+                
+            # Brief sleep to avoid busy waiting
+            time_module.sleep(0.1)
+        
+        # Return whatever completions we have
+        with self._spawn_lock:
+            return [e for e in self._completion_events if e.agent_id in target_agents]
+    
+    def _handle_sub_agent_completion(self, event: Event) -> None:
+        """Internal handler for sub-agent completion events."""
+        if event.data.get("parent_id") != self._team_id:
+            return  # Not for this team
+        
+        agent_id = event.data.get("agent_id")
+        if not agent_id:
+            return
+        
+        # Call registered completion callback if any
+        with self._spawn_lock:
+            callback = self._completion_callbacks.get(agent_id)
+            if callback:
+                try:
+                    completion_event = SubAgentCompletionEvent(
+                        agent_id=agent_id,
+                        task_id=event.data.get("task_id", ""),
+                        result=event.data.get("result"),
+                        success=event.data.get("success", False),
+                        error=event.data.get("error"),
+                        completion_time=event.data.get("completion_time", time.time()),
+                        parent_id=event.data.get("parent_id"),
+                        metadata=event.data.get("metadata", {})
+                    )
+                    callback(completion_event)
+                except Exception as e:
+                    logger.warning(f"Completion callback failed for agent {agent_id}: {e}")
+            
+            # Clean up completed agent
+            if agent_id in self._spawned_agents:
+                del self._spawned_agents[agent_id]
+            if agent_id in self._completion_callbacks:
+                del self._completion_callbacks[agent_id]
+
+    # Async variants for asyncio-based orchestration (per AGENTS.md guidelines)
+    
+    async def aspawn_sub_agent(
+        self,
+        agent: Agent,
+        task: Any,
+        completion_callback: Optional[Callable[[SubAgentCompletionEvent], Any]] = None,
+        metadata: Optional[Dict[str, Any]] = None
+    ) -> SpawnedSubAgent:
+        """Async version of spawn_sub_agent using asyncio primitives."""
+        import asyncio
+        
+        # Use asyncio lock for async coordination
+        if not hasattr(self, '_async_spawn_lock'):
+            self._async_spawn_lock = asyncio.Lock()
+        
+        async with self._async_spawn_lock:
+            # Create unique IDs
+            agent_id = f"{self._team_id}_{str(uuid.uuid4())[:8]}"
+            task_id = f"task_{str(uuid.uuid4())[:8]}"
+            
+            # Ensure event bus is initialized
+            if self._event_bus is None:
+                self._event_bus = EventBus()
+                # Subscribe to completion events for this team
+                self._event_bus.subscribe(
+                    self._handle_sub_agent_completion, 
+                    [EventType.SUBAGENT_COMPLETED.value, EventType.SUBAGENT_ERROR.value]
+                )
+            
+            # Create spawned agent record
+            spawned = SpawnedSubAgent(
+                agent_id=agent_id,
+                task_id=task_id,
+                agent=agent,
+                task=task,
+                spawn_time=time.time(),
+                parent_id=self._team_id,
+                metadata=metadata or {}
+            )
+            
+            # Store spawn record and callback
+            self._spawned_agents[agent_id] = spawned
+            if completion_callback:
+                self._completion_callbacks[agent_id] = completion_callback
+            
+            # Publish spawn event
+            if self._event_bus:
+                self._event_bus.publish(
+                    EventType.SUBAGENT_SPAWNED,
+                    {
+                        "agent_id": agent_id,
+                        "task_id": task_id,
+                        "parent_id": self._team_id,
+                        "spawn_time": spawned.spawn_time,
+                        "metadata": spawned.metadata
+                    }
+                )
+        
+        # Execute sub-agent asynchronously using asyncio.create_task
+        async def _aexecute_sub_agent():
+            try:
+                # Use agent's async start method
+                result = await agent.astart(str(task))
+                await self.aannounce_completion(agent_id, task_id, result, success=True)
+            except Exception as e:
+                logger.warning(f"Sub-agent {agent_id} failed: {e}")
+                await self.aannounce_completion(agent_id, task_id, None, success=False, error=str(e))
+        
+        # Create task for background execution
+        asyncio.create_task(_aexecute_sub_agent())
+        
+        logger.debug(f"Async spawned sub-agent {agent_id} for task {task_id}")
+        return spawned
+
+    async def aannounce_completion(
+        self,
+        agent_id: str,
+        task_id: str,
+        result: Any,
+        success: bool = True,
+        error: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None
+    ) -> None:
+        """Async version of announce_completion."""
+        import asyncio
+        
+        if not hasattr(self, '_async_spawn_lock'):
+            self._async_spawn_lock = asyncio.Lock()
+        
+        async with self._async_spawn_lock:
+            # Create completion event
+            completion_event = SubAgentCompletionEvent(
+                agent_id=agent_id,
+                task_id=task_id,
+                result=result,
+                success=success,
+                error=error,
+                parent_id=self._team_id,
+                metadata=metadata or {}
+            )
+            
+            # Store the completion event
+            self._completion_events.append(completion_event)
+            
+            # Publish completion event via event bus (use async if available)
+            if self._event_bus:
+                event_type = EventType.SUBAGENT_COMPLETED.value if success else EventType.SUBAGENT_ERROR.value
+                if hasattr(self._event_bus, 'publish_async'):
+                    await self._event_bus.publish_async(
+                        event_type,
+                        {
+                            "agent_id": agent_id,
+                            "task_id": task_id,
+                            "parent_id": self._team_id,
+                            "result": result,
+                            "success": success,
+                            "error": error,
+                            "completion_time": completion_event.completion_time,
+                            "metadata": completion_event.metadata
+                        }
+                    )
+                else:
+                    # Fallback to sync publish
+                    self._event_bus.publish(
+                        event_type,
+                        {
+                            "agent_id": agent_id,
+                            "task_id": task_id,
+                            "parent_id": self._team_id,
+                            "result": result,
+                            "success": success,
+                            "error": error,
+                            "completion_time": completion_event.completion_time,
+                            "metadata": completion_event.metadata
+                        }
+                    )
+            
+            logger.debug(f"Async announced completion for sub-agent {agent_id}: {'success' if success else 'error'}")
+
+    async def await_for_completions(
+        self,
+        timeout: Optional[float] = None,
+        agent_ids: Optional[List[str]] = None
+    ) -> List[SubAgentCompletionEvent]:
+        """Async, event-driven version of wait_for_completions."""
+        import asyncio
+        
+        # Read target agents inside lock to avoid race condition
+        if not hasattr(self, '_async_spawn_lock'):
+            self._async_spawn_lock = asyncio.Lock()
+        
+        async with self._async_spawn_lock:
+            target_agents = agent_ids or list(self._spawned_agents.keys())
+        
+        if not target_agents:
+            # Return existing completions for the requested agents
+            async with self._async_spawn_lock:
+                return [e for e in self._completion_events if not agent_ids or e.agent_id in agent_ids]
+        
+        # Use asyncio.Event for efficient waiting
+        completion_event = asyncio.Event()
+        completed_agents = set()
+        
+        def check_completions():
+            """Check if all target agents are completed - THREAD-SAFE version."""
+            for event in self._completion_events:
+                if event.agent_id in target_agents:
+                    completed_agents.add(event.agent_id)
+            
+            if completed_agents >= set(target_agents):
+                # CRITICAL FIX: Use call_soon_threadsafe() to safely set event from background thread
+                try:
+                    loop = asyncio.get_running_loop()
+                    loop.call_soon_threadsafe(completion_event.set)
+                except RuntimeError:
+                    # Fallback if no event loop running
+                    pass
+        
+        # Check initial state
+        check_completions()
+        
+        if not completion_event.is_set():
+            # Subscribe to completion events temporarily
+            def completion_handler(event):
+                if event.data.get("parent_id") == self._team_id:
+                    check_completions()
+            
+            subscriber_id = self._event_bus.subscribe(
+                completion_handler,
+                [EventType.SUBAGENT_COMPLETED.value, EventType.SUBAGENT_ERROR.value]
+            ) if self._event_bus else None
+            
+            try:
+                # Wait for completion or timeout
+                await asyncio.wait_for(completion_event.wait(), timeout=timeout)
+            except asyncio.TimeoutError:
+                pass
+            finally:
+                # Unsubscribe
+                if subscriber_id and self._event_bus:
+                    self._event_bus.unsubscribe(subscriber_id)
+        
+        # Return completed events
+        async with self._async_spawn_lock:
+            return [e for e in self._completion_events if e.agent_id in target_agents]
+
     def __enter__(self):
         """Context manager entry point for resource management."""
         return self