fix(examples): phase-2 validation fixes for memory/knowledge API and token metrics (#1859)

* fix: correct TypeError in 3 example files by using proper parameter names

- Fix MemoryConfig(memory=...) → user_id=... in agent_centric_api.py
- Fix Session(memory=...) → user_id=... in session-example.py
- Fix build_context_for_task(memory=...) → user_id=... in memory-quality-example.py

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

* fix(examples): align AgentTeam memory and knowledge configs with v4.6.52 API

---------

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

README.md

@@ -382,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

@@ -911,6 +911,18 @@ Methods:
 * <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>

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.48" \
+    "praisonai>=4.6.52" \
     "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.48" \
+    "praisonai>=4.6.52" \
     "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.48" \
+    "praisonai>=4.6.52" \
     "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,12 @@ 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={
-        "vector_store": "qdrant",
-        "url": "http://localhost:6333"
+        "sources": [doc_path],
+        "vector_store": {
+            "provider": "qdrant",
+            "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/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,13 +105,13 @@ def main():
         agents=[researcher, retriever],
         tasks=[store_task, verify_task, query_task, query_both_task],  # Use same verbose level as memory
         memory={
-            "provider": "rag",
             "embedder": {
                 "provider": "openai",
                 "config": {
                     "model": "text-embedding-3-small"
                 }
-            }
+            },
+            "config": {"provider": "sqlite"},
         }
     )
 

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

@@ -12,8 +12,8 @@ print("=== Advanced Graph Memory Integration Example ===\n")
 
 # Configure simple graph memory
 memory_config = {
-    "provider": "mem0",
     "config": {
+        "provider": "mem0",
         "vector_store": {
             "provider": "chroma",
             "config": {
@@ -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

@@ -12,8 +12,8 @@ print("=== Advanced Memory Search and Optimization Example ===\n")
 
 # Configure optimized memory settings
 memory_config = {
-    "provider": "mem0",
     "config": {
+        "provider": "mem0",
         "vector_store": {
             "provider": "chroma",
             "config": {
@@ -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,13 +86,13 @@ def main():
         agents=[researcher, retriever],
         tasks=[store_task, retrieve_task],
         memory={
-            "provider": "rag",
             "embedder": {
                 "provider": "gemini",
                 "config": {
                     "model": "text-embedding-004",
                 }
-            }
+            },
+            "config": {"provider": "sqlite"},
         }
     )
     

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/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,12 @@ 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={
+        "sources": ["kag-research-paper.pdf"],
+        "vector_store": config["vector_store"],
+        "embedder": config["embedder"],
+    },
+    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

@@ -28,8 +28,8 @@ if not os.environ.get("OPENAI_API_KEY"):
 def main():
     # MongoDB memory configuration
     mongodb_memory_config = {
-        "provider": "mongodb",
         "config": {
+            "provider": "mongodb",
             "connection_string": "mongodb://localhost:27017/",  # Replace with your MongoDB connection string
             "database": "praisonai_memory",
             "use_vector_search": True,  # Enable Atlas Vector Search (requires MongoDB Atlas)
@@ -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}")

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"
+        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/token-metrics/01_basic_token_tracking.py

@@ -14,7 +14,7 @@ def main():
         goal="Find information about a topic",
         backstory="You are an expert researcher who finds comprehensive information.",
         llm="gpt-4o-mini",  # Use a cost-effective model for examples
-        metrics=True  # 🎯 SIMPLIFIED: Just add this parameter!
+        output={"metrics": True}  # Token metrics via consolidated output config
     )
     
     # Create a simple task

examples/python/token-metrics/02_session_metrics.py

@@ -14,7 +14,7 @@ def main():
         goal="Analyze data and provide insights",
         backstory="You are an experienced data analyst with expertise in market research.",
         llm="gpt-4o-mini",
-        metrics=True  # 🎯 Enable metrics tracking
+        output={"metrics": True}
     )
     
     writer_agent = Agent(
@@ -23,7 +23,7 @@ def main():
         goal="Create well-structured content",
         backstory="You are a skilled technical writer who creates clear, engaging content.",
         llm="gpt-4o-mini",
-        metrics=True  # 🎯 Enable metrics tracking
+        output={"metrics": True}
     )
     
     # Create multiple tasks

examples/python/token-metrics/03_cost_estimation.py

@@ -14,7 +14,7 @@ def main():
         goal="Conduct comprehensive market research",
         backstory="You are a seasoned market researcher with 10+ years of experience.",
         llm="gpt-4o-mini",
-        metrics=True  # 🎯 Enable automatic metrics tracking
+        output={"metrics": True}
     )
     
     strategy_agent = Agent(
@@ -23,7 +23,7 @@ def main():
         goal="Develop actionable business strategies",
         backstory="You are a strategic consultant who translates research into business plans.",
         llm="gpt-4o-mini",
-        metrics=True  # 🎯 Enable automatic metrics tracking
+        output={"metrics": True}
     )
     
     # Create tasks that will generate substantial token usage

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/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:

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/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.

src/praisonai-agents/.gitignore

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

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/praisonaiagents/__init__.py

@@ -384,6 +384,7 @@ _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'),
     
     # Parameter resolver
     'resolve': ('praisonaiagents.config.param_resolver', 'resolve'),

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

@@ -556,6 +556,7 @@ class Agent(SandboxMixin, UnifiedExecutionMixin, ToolExecutionMixin, ChatHandler
         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
         sandbox: Optional[Union[bool, 'SandboxConfig']] = None,  # Sandbox for safe code execution
     ):
         """Initialize an Agent instance.
@@ -660,6 +661,14 @@ class Agent(SandboxMixin, UnifiedExecutionMixin, ToolExecutionMixin, ChatHandler
                 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.
@@ -1430,6 +1439,34 @@ class Agent(SandboxMixin, UnifiedExecutionMixin, ToolExecutionMixin, ChatHandler
             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
         # ============================================================
@@ -2026,6 +2063,7 @@ Your Goal: {self.goal}
             '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
             'tool_timeout': getattr(self, '_tool_timeout', None),

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

@@ -331,6 +331,7 @@ Your Goal: {self.goal}"""
         - Callable functions
         - String function names
         - Objects with to_openai_tool() method
+        - Tool Search progressive disclosure (if enabled)
         
         Args:
             tools: List of tools in various formats or None to use self.tools
@@ -344,10 +345,14 @@ Your Goal: {self.goal}"""
         if not tools:
             return []
         
-        # Check cache first
+        # Check cache first - include tool_search config in cache key for safety
         tools_key = self._get_tools_cache_key(tools)
-        cached_tools = self._cache_get(self._formatted_tools_cache, tools_key)
-        if cached_tools is not None:
+        tool_search_enabled = getattr(self, '_tool_search_config', None) is not None
+        cache_key = f"{tools_key}:tool_search={tool_search_enabled}"
+        cached_entry = self._cache_get(self._formatted_tools_cache, cache_key)
+        if cached_entry is not None:
+            cached_tools, cached_metadata = cached_entry
+            self._tool_search_metadata = cached_metadata
             return cached_tools
             
         formatted_tools = []
@@ -399,9 +404,46 @@ Your Goal: {self.goal}"""
                 logging.error(f"Tools are not JSON serializable: {e}")
                 return []
         
-        # Cache the formatted tools with LRU eviction
-        self._cache_put(self._formatted_tools_cache, tools_key, formatted_tools)
-        return formatted_tools
+        # Apply tool search assembly if enabled (after formatting, before caching)
+        if hasattr(self, '_tool_search_config') and self._tool_search_config is not None:
+            try:
+                from ..tools.tool_search import assemble_tool_defs
+                # Get context length from LLM config if available
+                context_length = getattr(self, '_context_window_size', None)
+                
+                # Assemble tools with bridge mode check
+                assembled_tools, metadata = assemble_tool_defs(
+                    tool_defs=formatted_tools,
+                    config=self._tool_search_config,
+                    context_length=context_length
+                )
+                
+                # Store metadata for bridge tool dispatch
+                self._tool_search_metadata = metadata
+                formatted_tools = assembled_tools
+                
+            except ImportError:
+                # Tool search module not available, continue with original tools
+                logging.warning("Tool search requested but tool_search module not available")
+        
+        # Strip __praisonai_deferrable__ from provider-facing tool payloads
+        # Keep the marker only for internal tool classification
+        cleaned_tools = []
+        for tool in formatted_tools:
+            if isinstance(tool, dict) and "__praisonai_deferrable__" in tool:
+                tool_copy = tool.copy()
+                tool_copy.pop("__praisonai_deferrable__", None)
+                cleaned_tools.append(tool_copy)
+            else:
+                cleaned_tools.append(tool)
+        
+        # Cache the formatted tools with LRU eviction, including tool search metadata
+        self._cache_put(
+            self._formatted_tools_cache,
+            cache_key,
+            (cleaned_tools, getattr(self, "_tool_search_metadata", None)),
+        )
+        return cleaned_tools
 
     def _build_multimodal_prompt(
         self, 
@@ -507,7 +549,7 @@ Your Goal: {self.goal}"""
             emit_events=True
         )
 
-    def _chat_completion(self, messages, temperature=1.0, tools=None, stream=True, reasoning_steps=False, task_name=None, task_description=None, task_id=None, response_format=None, _retry_depth=0):
+    def _chat_completion(self, messages, temperature=1.0, tools=None, stream=None, reasoning_steps=False, task_name=None, task_description=None, task_id=None, response_format=None, _retry_depth=0):
         start_time = time.time()
 
         # --- Context compaction (opt-in via ExecutionConfig.context_compaction) ---
@@ -577,6 +619,39 @@ Your Goal: {self.goal}"""
         # Use the new _format_tools_for_completion helper method
         formatted_tools = self._format_tools_for_completion(tools)
 
+        # Smart fallback for streaming: try streaming first, fall back to non-streaming if unsupported
+        if stream is None:
+            # Auto-detect: prefer streaming for better UX, fallback if adapter doesn't support it
+            try:
+                # First attempt: try with streaming enabled for better user experience
+                stream_callback = self.stream_emitter.emit if hasattr(self, 'stream_emitter') else None
+                final_response = self._execute_unified_chat_completion(
+                    messages=messages,
+                    temperature=temperature,
+                    tools=formatted_tools,
+                    stream=True,  # Try streaming first
+                    reasoning_steps=reasoning_steps,
+                    task_name=task_name,
+                    task_description=task_description,
+                    task_id=task_id,
+                    response_format=response_format,
+                    stream_callback=stream_callback,
+                    emit_events=True
+                )
+                return final_response
+            except ValueError as e:
+                if "Streaming is not supported" in str(e):
+                    # Fallback: retry with non-streaming for sync adapters
+                    logging.debug(f"{self.name}: Streaming not supported by adapter, falling back to non-streaming")
+                    stream = False  # Set for the main execution below
+                else:
+                    raise  # Re-raise if it's a different ValueError
+            except Exception:
+                # For any other exception, fall back to non-streaming
+                logging.debug(f"{self.name}: Streaming attempt failed, falling back to non-streaming")
+                stream = False  # Set for the main execution below
+        
+        # If stream was explicitly set or fallback occurred, use the specified/fallback value
         try:
             # NEW: Unified protocol dispatch path (Issue #1304, #1362)
             # UNIFIED: Single protocol-driven dispatch path (fixes DRY violation)
@@ -774,7 +849,7 @@ Your Goal: {self.goal}"""
         messages, 
         temperature=1.0, 
         tools=None, 
-        stream=True,
+        stream=None,
         reasoning_steps=False,
         task_name=None,
         task_description=None,
@@ -806,6 +881,47 @@ Your Goal: {self.goal}"""
             # Cache the dispatcher
             self._unified_dispatcher = dispatcher
         
+        # Smart fallback for streaming: try streaming first, fall back to non-streaming if unsupported
+        if stream is None:
+            # Auto-detect: prefer streaming for better UX, fallback if adapter doesn't support it
+            try:
+                # First attempt: try with streaming enabled for better user experience
+                if stream_callback is None and hasattr(self, 'stream_emitter'):
+                    stream_callback = getattr(self.stream_emitter, 'emit', None)
+                final_response = self._unified_dispatcher.chat_completion(
+                    messages=messages,
+                    tools=tools,
+                    tool_choice=getattr(self, 'tool_choice', None),
+                    temperature=temperature,
+                    max_tokens=getattr(self, 'max_tokens', None),
+                    stream=True,  # Try streaming first
+                    response_format=response_format,
+                    execute_tool_fn=getattr(self, 'execute_tool', None),
+                    console=self.console if (self.verbose or True) else None,  # Enable console for streaming
+                    display_fn=self._display_generating if self.verbose else None,
+                    stream_callback=stream_callback,
+                    emit_events=emit_events,
+                    verbose=self.verbose,
+                    max_iterations=10,
+                    reasoning_steps=reasoning_steps,
+                    task_name=task_name,
+                    task_description=task_description,
+                    task_id=task_id,
+                    agent_name=getattr(self, 'name', 'assistant')
+                )
+                return final_response
+            except ValueError as e:
+                if "Streaming is not supported" in str(e):
+                    # Fallback: retry with non-streaming for sync adapters
+                    logging.debug(f"Agent: Streaming not supported by adapter, falling back to non-streaming")
+                    stream = False  # Set for the main execution below
+                else:
+                    raise  # Re-raise if it's a different ValueError
+            except Exception:
+                # For any other exception, fall back to non-streaming
+                logging.debug(f"Agent: Streaming attempt failed, falling back to non-streaming")
+                stream = False  # Set for the main execution below
+
         # Execute unified dispatch with all necessary parameters
         # Includes all parameters from both legacy paths to ensure full compatibility
         try:
@@ -857,7 +973,7 @@ Your Goal: {self.goal}"""
         messages, 
         temperature=1.0, 
         tools=None, 
-        stream=True,
+        stream=True,  # Async methods keep stream=True default (async adapters support streaming vs sync smart fallback)
         reasoning_steps=False,
         task_name=None,
         task_description=None,

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

@@ -135,6 +135,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
         
@@ -954,3 +960,97 @@ 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}"

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

@@ -39,6 +39,7 @@ __all__ = [
     "CachingConfig",
     "HooksConfig",
     "SkillsConfig",
+    "ToolSearchConfig",
     # Type aliases
     "MemoryParam",
     "KnowledgeParam",
@@ -122,6 +123,7 @@ _MODULE_MAP = {
     "CachingConfig": "feature_configs",
     "HooksConfig": "feature_configs",
     "SkillsConfig": "feature_configs",
+    "ToolSearchConfig": "feature_configs",
     "MemoryParam": "feature_configs",
     "KnowledgeParam": "feature_configs",
     "PlanningParam": "feature_configs",

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

@@ -8,6 +8,7 @@ Provides dataclasses for consolidated feature configuration:
 - ReflectionConfig: Self-reflection settings
 - GuardrailConfig: Safety and validation
 - WebConfig: Web search and fetch
+- ToolSearchConfig: Progressive tool disclosure
 
 All configs follow the agent-centric pattern:
 - False: Feature disabled (zero overhead)
@@ -16,21 +17,22 @@ All configs follow the agent-centric pattern:
 - Instance: Pre-configured manager/engine
 
 Usage:
-    from praisonaiagents import Agent, MemoryConfig, KnowledgeConfig
+    from praisonaiagents import Agent, MemoryConfig, KnowledgeConfig, ToolSearchConfig
     
     # Simple enable
-    agent = Agent(instructions="...", memory=True)
+    agent = Agent(instructions="...", memory=True, tool_search=True)
     
     # With config
     agent = Agent(
         instructions="...",
         memory=MemoryConfig(backend="redis", user_id="user123"),
         knowledge=KnowledgeConfig(sources=["docs/"], rerank=True),
+        tool_search=ToolSearchConfig(enabled="auto", threshold_pct=15),
     )
 """
 
 from dataclasses import dataclass, field
-from typing import Dict, List, Any, Optional, Callable, Tuple, Union
+from typing import Dict, List, Any, Optional, Callable, Tuple, Union, FrozenSet
 from enum import Enum
 
 # Import AutonomyConfig from canonical location (no circular dep)
@@ -1072,6 +1074,25 @@ class MultiAgentMemoryConfig:
         }
 
 
+# Import ToolSearchConfig from tools module to avoid duplication
+def __get_tool_search_config():
+    try:
+        from ..tools.tool_search import ToolSearchConfig as _ToolSearchConfig
+        return _ToolSearchConfig
+    except ImportError:
+        # Fallback minimal config if tools module not available
+        @dataclass
+        class FallbackToolSearchConfig:
+            enabled: Union[bool, str] = "auto"
+            threshold_pct: float = 10.0
+            search_default_limit: int = 5
+            max_search_limit: int = 20
+            core_tools: Optional[FrozenSet[str]] = None
+        return FallbackToolSearchConfig
+
+ToolSearchConfig = __get_tool_search_config()
+
+
 class AutonomyLevel(str, Enum):
     """Autonomy levels for agent behavior."""
     SUGGEST = "suggest"
@@ -1092,6 +1113,7 @@ CachingParam = Union[bool, CachingConfig]
 HooksParam = Union[List[Any], HooksConfig]
 SkillsParam = Union[List[str], SkillsConfig]
 AutonomyParam = Union[bool, Dict[str, Any], "AutonomyConfig"]
+ToolSearchParam = Union[bool, str, Dict[str, Any], ToolSearchConfig]
 
 
 # =============================================================================
@@ -1343,6 +1365,21 @@ def resolve_autonomy(value: AutonomyParam) -> Optional[AutonomyConfig]:
     return value
 
 
+def resolve_tool_search(value: ToolSearchParam) -> Optional[ToolSearchConfig]:
+    """Resolve tool_search= parameter following precedence ladder."""
+    if value is None or value is False:
+        return None
+    if value is True:
+        return ToolSearchConfig()
+    if isinstance(value, str):
+        return ToolSearchConfig(enabled=value)
+    if isinstance(value, dict):
+        return ToolSearchConfig(**value)
+    if isinstance(value, ToolSearchConfig):
+        return value
+    return value
+
+
 __all__ = [
     # Enums
     "MemoryBackend",
@@ -1368,6 +1405,7 @@ __all__ = [
     "HooksConfig",
     "SkillsConfig",
     "AutonomyConfig",
+    "ToolSearchConfig",
     # Config classes (Multi-Agent)
     "MultiAgentHooksConfig",
     "MultiAgentOutputConfig",
@@ -1388,6 +1426,7 @@ __all__ = [
     "HooksParam",
     "SkillsParam",
     "AutonomyParam",
+    "ToolSearchParam",
     # Precedence ladder resolvers
     "resolve_memory",
     "resolve_knowledge",
@@ -1399,4 +1438,5 @@ __all__ = [
     "resolve_execution",
     "resolve_caching",
     "resolve_autonomy",
+    "resolve_tool_search",
 ]

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

@@ -83,11 +83,23 @@ def __getattr__(name):
     elif name == "OAUTH_CALLBACK_PATH":
         from .mcp_oauth_callback import OAUTH_CALLBACK_PATH
         return OAUTH_CALLBACK_PATH
+    # Protocols (lazy loaded)
+    elif name == "MCPClientProtocol":
+        from .protocols import MCPClientProtocol
+        return MCPClientProtocol
+    # Loader (lazy loaded)
+    elif name == "load_mcp_tools":
+        from .loader import load_mcp_tools
+        return load_mcp_tools
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
 __all__ = [
     # Client
     "MCP",
+    # Protocols (lazy loaded)
+    "MCPClientProtocol",
+    # Loader (lazy loaded)
+    "load_mcp_tools",
     # Server (lazy loaded)
     "ToolsMCPServer",
     "launch_tools_mcp_server",

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

@@ -0,0 +1,92 @@
+"""
+MCP Tools Loader for PraisonAI Agents.
+
+Provides a thin helper function to load MCP tools for agents
+following the agent-centric design principles in AGENTS.md.
+"""
+
+from typing import List, Optional
+
+from ..memory.mcp_config import MCPConfig, MCPConfigManager
+from .mcp import MCP
+
+
+def load_mcp_tools(
+    names: Optional[List[str]] = None,
+    *,
+    configs: Optional[List[MCPConfig]] = None,
+    prefix_tools: bool = True,
+) -> List[MCP]:
+    """
+    Load MCP tools for use with agents.
+    
+    This is a thin helper that bridges the gap between configuration
+    and agent tool setup, following the "few lines to run an agent"
+    principle from AGENTS.md.
+    
+    Args:
+        names: Specific config names to load (None = all enabled)
+        configs: Optional injected configs from wrapper TOML loader
+        prefix_tools: Whether to prefix tool names when multiple servers
+            are loaded to avoid collisions (default: True).
+            NOTE: Currently not implemented - collisions possible with multiple servers.
+    
+    Returns:
+        List of MCP client instances ready for agent use
+    
+    Examples:
+        # Load all enabled MCP servers from JSON config
+        >>> from praisonaiagents import Agent
+        >>> from praisonaiagents.mcp import load_mcp_tools
+        >>> agent = Agent(name="assistant", tools=load_mcp_tools())
+        
+        # Load specific servers only
+        >>> tools = load_mcp_tools(["filesystem", "github"])
+        >>> agent = Agent(name="coder", tools=tools)
+        
+        # Inject configs from wrapper TOML
+        >>> from praisonai.cli.configuration import get_config_loader
+        >>> loader = get_config_loader()
+        >>> config = loader.load()
+        >>> toml_configs = [MCPConfig(...) for server in config.mcp.servers]
+        >>> tools = load_mcp_tools(configs=toml_configs)
+    """
+    mcp_clients: List[MCP] = []
+    
+    if configs is not None:
+        # Use injected configs from wrapper (TOML loading)
+        target_configs = configs
+        if names:
+            # Filter by names if specified
+            target_configs = [c for c in configs if c.name in names]
+    else:
+        # Load from MCPConfigManager (JSON files)
+        manager = MCPConfigManager()
+        if names:
+            target_configs = [manager.get_config(name) for name in names]
+            target_configs = [c for c in target_configs if c and c.enabled]
+        else:
+            target_configs = manager.get_enabled_configs()
+    
+    for config in target_configs:
+        if not config.enabled:
+            continue
+            
+        # Convert config to MCP instance
+        mcp = config.to_mcp_instance()
+        if mcp:
+            # Apply tool filtering if configured (B3)
+            # TODO: Wire include/exclude filters when B3 is implemented
+            
+            # Apply tool prefix if multiple servers (B4)
+            if prefix_tools and len(target_configs) > 1:
+                # Sanitize server name for use as prefix
+                server_name = config.name.replace('-', '_').replace(' ', '_')
+                # TODO: Tool name prefixing not yet implemented
+                # Need to modify tool names in MCP instance to avoid collisions
+                # e.g., "read_file" -> "filesystem_read_file"
+                pass
+            
+            mcp_clients.append(mcp)
+    
+    return mcp_clients

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

@@ -29,8 +29,8 @@ class MCPToolRunner(threading.Thread):
         super().__init__(daemon=True)
         self.server_params = server_params
         self.queue = queue.Queue()
-        self.result_queue = queue.Queue()
         self.initialized = threading.Event()
+        self._init_error = None
         self.tools = []
         self.timeout = timeout
         self._tool_timings = {}
@@ -66,12 +66,12 @@ class MCPToolRunner(threading.Thread):
                                 if item is None:  # Shutdown signal
                                     break
                                 
-                                tool_name, arguments = item
+                                response_queue, tool_name, arguments = item
                                 try:
                                     result = await session.call_tool(tool_name, arguments)
-                                    self.result_queue.put((True, result))
+                                    response_queue.put((True, result))
                                 except Exception as e:
-                                    self.result_queue.put((False, str(e)))
+                                    response_queue.put((False, str(e)))
                             except queue.Empty:
                                 pass
                             
@@ -80,8 +80,8 @@ class MCPToolRunner(threading.Thread):
                         except asyncio.CancelledError:
                             break
         except Exception as e:
+            self._init_error = f"MCP initialization error: {str(e)}"
             self.initialized.set()  # Ensure we don't hang
-            self.result_queue.put((False, f"MCP initialization error: {str(e)}"))
     
     def call_tool(self, tool_name, arguments):
         """Call an MCP tool and wait for the result."""
@@ -100,16 +100,25 @@ class MCPToolRunner(threading.Thread):
                 if telemetry:
                     telemetry.track_tool_usage(tool_name, success=False, execution_time=0)
                 return f"Error: MCP initialization timed out after {self.timeout} seconds"
+
+        if self._init_error:
+            if telemetry:
+                telemetry.track_tool_usage(tool_name, success=False, execution_time=0)
+            return f"Error: {self._init_error}"
         
         # Start timing after initialization check
         start_time = time.time()
         is_success = False
+        response_queue = queue.Queue(maxsize=1)
         try:
-            # Put request in queue
-            self.queue.put((tool_name, arguments))
+            # Put request in queue with caller-specific response channel
+            self.queue.put((response_queue, tool_name, arguments))
             
-            # Wait for result
-            success, result = self.result_queue.get()
+            # Wait for result with timeout
+            try:
+                success, result = response_queue.get(timeout=self.timeout)
+            except queue.Empty:
+                return f"Error: MCP tool call timed out after {self.timeout} seconds"
             if not success:
                 return f"Error: {result}"
             
@@ -187,7 +196,8 @@ class MCP:
         ```
     """
     
-    def __init__(self, command_or_string=None, args=None, *, command=None, timeout=60, debug=False, **kwargs):
+    def __init__(self, command_or_string=None, args=None, *, command=None, timeout=60, debug=False, 
+                 allowed_tools: Optional[List[str]] = None, disabled_tools: Optional[List[str]] = None, **kwargs):
         """
         Initialize the MCP connection and get tools.
         
@@ -201,7 +211,13 @@ class MCP:
             command: Alternative parameter name for backward compatibility
             timeout: Timeout in seconds for MCP server initialization and tool calls (default: 60)
             debug: Enable debug logging for MCP operations (default: False)
+            allowed_tools: Include whitelist - only these tools will be available (default: None = all tools)
+            disabled_tools: Exclude blacklist - these tools will be filtered out (default: None = no exclusions)
             **kwargs: Additional parameters for StdioServerParameters
+            
+        Note:
+            If both allowed_tools and disabled_tools are specified, allowed_tools takes precedence
+            (include wins over exclude).
         """
         # Check if MCP is available
         if not MCP_AVAILABLE:
@@ -240,6 +256,8 @@ class MCP:
         # Store additional parameters
         self.timeout = timeout
         self.debug = debug
+        self.allowed_tools = allowed_tools
+        self.disabled_tools = disabled_tools
         
         # Check if this is a WebSocket URL (ws:// or wss://)
         if isinstance(command_or_string, str) and re.match(r'^wss?://', command_or_string):
@@ -254,7 +272,7 @@ class MCP:
                 auth_token=auth_token,
                 options=kwargs
             )
-            self._tools = list(self.websocket_client.tools)
+            self._tools = self._apply_tool_filters(list(self.websocket_client.tools))
             self.is_sse = False
             self.is_http_stream = False
             self.is_websocket = True
@@ -268,7 +286,7 @@ class MCP:
                 # Legacy SSE URL - use SSE transport for backward compatibility
                 from .mcp_sse import SSEMCPClient
                 self.sse_client = SSEMCPClient(command_or_string, debug=debug, timeout=timeout)
-                self._tools = list(self.sse_client.tools)
+                self._tools = self._apply_tool_filters(list(self.sse_client.tools))
                 self.is_sse = True
                 self.is_http_stream = False
                 self.is_npx = False
@@ -295,7 +313,7 @@ class MCP:
                     timeout=timeout,
                     options=transport_options
                 )
-                self._tools = list(self.http_stream_client.tools)
+                self._tools = self._apply_tool_filters(list(self.http_stream_client.tools))
                 self.is_sse = False
                 self.is_http_stream = True
                 self.is_npx = False
@@ -325,20 +343,10 @@ class MCP:
         self.is_sse = False
         self.is_http_stream = False
         
-        # Ensure UTF-8 encoding in environment for Docker compatibility
-        env = kwargs.get('env', {})
-        if not env:
-            env = os.environ.copy()
-        
-        # Always set Python encoding
-        env['PYTHONIOENCODING'] = 'utf-8'
-        
-        # Only set locale variables on Unix systems
-        if platform.system() != 'Windows':
-            env.update({
-                'LC_ALL': 'C.UTF-8',
-                'LANG': 'C.UTF-8'
-            })
+        # Build safe environment for stdio MCP servers
+        # Use safe baseline + explicit env from config (B5 security policy)
+        custom_env = kwargs.get('env', {})
+        env = self._build_safe_env(custom_env)
         
         kwargs['env'] = env
         
@@ -369,7 +377,7 @@ class MCP:
             self._initialize_npx_mcp_tools(cmd, arguments)
         else:
             # Generate tool functions immediately and store them
-            self._tools = self._generate_tool_functions()
+            self._tools = self._apply_tool_filters(self._generate_tool_functions())
     
     def _generate_tool_functions(self) -> List[Callable]:
         """
@@ -489,7 +497,7 @@ class MCP:
                 logging.debug(f"Initializing NPX MCP tools with command: {cmd} {' '.join(arguments)}")
             
             # Generate tool functions using the regular MCP approach
-            self._tools = self._generate_tool_functions()
+            self._tools = self._apply_tool_filters(self._generate_tool_functions())
             
             if self.debug:
                 logging.debug(f"Generated {len(self._tools)} NPX MCP tools")
@@ -499,6 +507,100 @@ class MCP:
                 logging.error(f"Failed to initialize NPX MCP tools: {e}")
             raise RuntimeError(f"Failed to initialize NPX MCP tools: {e}")
     
+    def _apply_tool_filters(self, raw_tools: List[Any]) -> List[Any]:
+        """
+        Apply tool filtering based on allowed_tools and disabled_tools.
+        
+        Args:
+            raw_tools: Unfiltered tool list
+            
+        Returns:
+            Filtered tool list
+        """
+        if not self.allowed_tools and not self.disabled_tools:
+            return raw_tools
+            
+        # Import filter functions
+        from .mcp_utils import filter_tools_by_allowlist, filter_disabled_tools
+        
+        # Convert tools to the format expected by filter functions (list of dicts with 'name')
+        # Raw tools might be function objects, so we need to extract names
+        tool_defs = []
+        for tool in raw_tools:
+            if hasattr(tool, '__name__'):
+                tool_defs.append({"name": tool.__name__, "_tool_obj": tool})
+            elif isinstance(tool, dict) and 'name' in tool:
+                tool_defs.append(tool)
+            else:
+                # Skip tools that don't have a clear name
+                continue
+        
+        # Include wins over exclude - apply allowlist exclusively if provided
+        if self.allowed_tools:
+            tool_defs = filter_tools_by_allowlist(tool_defs, self.allowed_tools)
+        elif self.disabled_tools:
+            tool_defs = filter_disabled_tools(tool_defs, self.disabled_tools)
+        
+        # Convert back to original format
+        filtered_tools = []
+        for tool_def in tool_defs:
+            if "_tool_obj" in tool_def:
+                filtered_tools.append(tool_def["_tool_obj"])
+            else:
+                filtered_tools.append(tool_def)
+        
+        return filtered_tools
+    
+    def _build_safe_env(self, custom_env: Optional[dict] = None) -> dict:
+        """
+        Build a safe environment for stdio MCP servers.
+        
+        Merges only safe baseline environment variables with explicit custom ones,
+        following the security policy outlined in AGENTS.md.
+        
+        Args:
+            custom_env: Optional custom environment variables from config
+            
+        Returns:
+            Safe environment dictionary
+        """
+        # Safe baseline environment variables
+        safe_baseline = {
+            'PATH': os.environ.get('PATH', ''),
+            'HOME': os.environ.get('HOME', os.path.expanduser('~')),
+            'USER': os.environ.get('USER', ''),
+            'LANG': os.environ.get('LANG', 'C.UTF-8'),
+            'LC_ALL': os.environ.get('LC_ALL', 'C.UTF-8'),
+            'PYTHONIOENCODING': 'utf-8',
+        }
+        
+        # Platform-specific safe variables
+        if platform.system() == 'Windows':
+            safe_baseline.update({
+                'SYSTEMROOT': os.environ.get('SYSTEMROOT', ''),
+                'COMSPEC': os.environ.get('COMSPEC', ''),
+                'USERNAME': os.environ.get('USERNAME', ''),
+                'USERPROFILE': os.environ.get('USERPROFILE', ''),
+                'APPDATA': os.environ.get('APPDATA', ''),
+                'LOCALAPPDATA': os.environ.get('LOCALAPPDATA', ''),
+                'TEMP': os.environ.get('TEMP', ''),
+                'TMP': os.environ.get('TMP', ''),
+            })
+        else:
+            safe_baseline.update({
+                'TMPDIR': os.environ.get('TMPDIR', '/tmp'),
+                'SHELL': os.environ.get('SHELL', '/bin/sh'),
+            })
+        
+        # Start with safe baseline
+        env = {k: v for k, v in safe_baseline.items() if v}  # Only include non-empty values
+        
+        # Add explicit custom environment variables from config
+        if custom_env:
+            env.update(custom_env)
+        
+        return env
+
     def __iter__(self) -> Iterable[Callable]:
         """
         Allow the MCP instance to be used directly as an iterable of tools.
@@ -609,7 +711,8 @@ class MCP:
                     "name": tool.name,
                     "description": tool.description if hasattr(tool, 'description') else f"Call the {tool.name} tool",
                     "parameters": parameters
-                }
+                },
+                "__praisonai_deferrable__": True  # Mark MCP tools as deferrable for tool search
             })
         
         return openai_tools

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

@@ -156,7 +156,8 @@ class HTTPStreamMCPTool:
                 "name": self.name,
                 "description": self.description,
                 "parameters": fixed_schema
-            }
+            },
+            "__praisonai_deferrable__": True  # Mark MCP tools as deferrable for tool search
         }
 
 class HTTPStreamTransport:

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

@@ -121,7 +121,8 @@ class SSEMCPTool:
                 "name": self.name,
                 "description": self.description,
                 "parameters": fixed_schema
-            }
+            },
+            "__praisonai_deferrable__": True  # Mark MCP tools as deferrable for tool search
         }
 
 class SSEMCPClient:

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

@@ -363,7 +363,8 @@ class WebSocketMCPTool:
                 "name": self.name,
                 "description": self.description,
                 "parameters": fixed_schema
-            }
+            },
+            "__praisonai_deferrable__": True  # Mark MCP tools as deferrable for tool search
         }
 
 class WebSocketMCPClient:

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

@@ -0,0 +1,110 @@
+"""
+Model Context Protocol (MCP) client protocols for PraisonAI Agents.
+
+Defines protocol interfaces for MCP client extensibility following
+the protocol-driven design pattern outlined in AGENTS.md.
+"""
+
+from typing import Any, Dict, List, Optional, Protocol, runtime_checkable
+
+
+@runtime_checkable
+class MCPClientProtocol(Protocol):
+    """
+    Protocol interface for MCP client implementations.
+    
+    Defines the minimal contract that all MCP clients must follow,
+    enabling extensibility through different transport mechanisms
+    (stdio, SSE, HTTP stream, WebSocket) while maintaining a
+    consistent interface for agents.
+    """
+
+    def list_tools(self) -> List[Dict[str, Any]]:
+        """
+        List available tools from the MCP server.
+        
+        Returns:
+            List of tool definitions from the server
+        """
+        ...
+
+    def get_tools(self) -> List[Dict[str, Any]]:
+        """
+        Get available tools from the MCP server.
+        
+        Alias for list_tools() to maintain backward compatibility.
+        
+        Returns:
+            List of tool definitions from the server
+        """
+        ...
+
+    def call_tool(self, name: str, args: Dict[str, Any]) -> Any:
+        """
+        Call a tool on the MCP server.
+        
+        Args:
+            name: Tool name to call
+            args: Tool arguments
+            
+        Returns:
+            Tool execution result
+            
+        Raises:
+            Exception: If tool call fails
+        """
+        ...
+
+    def shutdown(self) -> None:
+        """
+        Shutdown the MCP client and clean up resources.
+        
+        This should gracefully close connections, terminate
+        subprocesses, and release any held resources.
+        """
+        ...
+
+    async def async_list_tools(self) -> List[Dict[str, Any]]:
+        """
+        Asynchronously list available tools from the MCP server.
+        
+        Returns:
+            List of tool definitions from the server
+        """
+        ...
+
+    async def async_get_tools(self) -> List[Dict[str, Any]]:
+        """
+        Asynchronously get available tools from the MCP server.
+        
+        Alias for async_list_tools() to maintain backward compatibility.
+        
+        Returns:
+            List of tool definitions from the server
+        """
+        ...
+
+    async def async_call_tool(self, name: str, args: Dict[str, Any]) -> Any:
+        """
+        Asynchronously call a tool on the MCP server.
+        
+        Args:
+            name: Tool name to call
+            args: Tool arguments
+            
+        Returns:
+            Tool execution result
+            
+        Raises:
+            Exception: If tool call fails
+        """
+        ...
+
+    async def async_shutdown(self) -> None:
+        """
+        Asynchronously shutdown the MCP client and clean up resources.
+        
+        This should gracefully close connections, terminate
+        subprocesses, and release any held resources.
+        """
+        ...

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

@@ -81,7 +81,7 @@ class MCPConfig:
             MCP instance ready for use with agents
         """
         try:
-            from praisonaiagents.tools.mcp import MCP
+            from praisonaiagents.mcp import MCP
             
             # Resolve environment variables
             resolved_env = {}

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

@@ -144,6 +144,8 @@ class HierarchicalSessionStore(DefaultSessionStore):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._extended_cache: Dict[str, ExtendedSessionData] = {}
+        self._cache_mtimes: Dict[str, float] = {}  # Track file modification times
+
 
     def _load_session_from_disk(self, session_id: str, filepath: str) -> ExtendedSessionData:
         """Load extended session JSON from disk (caller must hold FileLock)."""
@@ -173,6 +175,43 @@ class HierarchicalSessionStore(DefaultSessionStore):
                 if isinstance(cached, ExtendedSessionData):
                     self._extended_cache[session_id] = cached
         return result
+
+    def _is_cache_valid(self, session_id: str) -> bool:
+        """Check if cached session is still valid based on file mtime."""
+        if session_id not in self._extended_cache:
+            return False
+        
+        filepath = self._get_session_path(session_id)
+        if not os.path.exists(filepath):
+            return False
+        
+        try:
+            current_mtime = os.path.getmtime(filepath)
+            cached_mtime = self._cache_mtimes.get(session_id, 0)
+            return current_mtime <= cached_mtime
+        except (OSError, IOError):
+            return False
+    
+    def _read_session_fresh(self, session_id: str) -> ExtendedSessionData:
+        """Reload from disk and keep _cache and _extended_cache in sync."""
+        session = super()._read_session_fresh(session_id)
+        if not isinstance(session, ExtendedSessionData):
+            session = ExtendedSessionData.from_session_data(session)
+            with self._lock:
+                self._cache[session_id] = session
+        
+        # Update cache with fresh file mtime
+        filepath = self._get_session_path(session_id)
+        try:
+            mtime = os.path.getmtime(filepath) if os.path.exists(filepath) else time.time()
+        except (OSError, IOError):
+            mtime = time.time()
+        
+        with self._lock:
+            self._extended_cache[session_id] = session
+            self._cache_mtimes[session_id] = mtime
+        
+        return session
     
     def add_message(
         self,
@@ -204,35 +243,14 @@ class HierarchicalSessionStore(DefaultSessionStore):
         )
     
     def _load_extended_session(self, session_id: str, force_reload: bool = False) -> ExtendedSessionData:
-        """Load extended session from disk."""
-        filepath = self._get_session_path(session_id)
-        
-        # Check cache first (unless force reload)
-        if not force_reload:
-            with self._lock:
-                if session_id in self._extended_cache:
-                    return self._extended_cache[session_id]
-        
-        # Load from disk
-        if not os.path.exists(filepath):
-            session = ExtendedSessionData(session_id=session_id)
-            with self._lock:
-                self._extended_cache[session_id] = session
-            return session
-        
-        with FileLock(filepath, self.lock_timeout):
-            try:
-                with open(filepath, "r", encoding="utf-8") as f:
-                    data = json.load(f)
-                session = ExtendedSessionData.from_dict(data)
-            except (json.JSONDecodeError, IOError) as e:
-                logger.warning(f"Failed to load session {session_id}: {e}")
-                session = ExtendedSessionData(session_id=session_id)
+        """Load extended session with smart caching based on file modification time."""
+        # Force reload bypasses cache validation
+        if force_reload or not self._is_cache_valid(session_id):
+            return self._read_session_fresh(session_id)
         
+        # Cache is valid, return cached version
         with self._lock:
-            self._extended_cache[session_id] = session
-        
-        return session
+            return self._extended_cache[session_id]
     
     def _save_extended_session(self, session: ExtendedSessionData) -> bool:
         """Save extended session to disk."""
@@ -260,8 +278,15 @@ class HierarchicalSessionStore(DefaultSessionStore):
                 
                 os.replace(temp_path, filepath)
                 
+                # Update cache with current file mtime after successful write
+                try:
+                    mtime = os.path.getmtime(filepath)
+                except (OSError, IOError):
+                    mtime = time.time()
+                
                 with self._lock:
                     self._extended_cache[session.session_id] = session
+                    self._cache_mtimes[session.session_id] = mtime
                 
                 return True
             except (IOError, OSError) as e:
@@ -304,12 +329,13 @@ class HierarchicalSessionStore(DefaultSessionStore):
             metadata=metadata or {},
         )
         
-        # Update parent's children list
+        # Update parent's children list without clobbering concurrent message writes
         if parent_id:
-            parent = self._load_extended_session(parent_id)
-            if sid not in parent.children_ids:
-                parent.children_ids.append(sid)
-                self._save_extended_session(parent)
+            def _apply(parent_session: SessionData) -> None:
+                assert isinstance(parent_session, ExtendedSessionData)
+                if sid not in parent_session.children_ids:
+                    parent_session.children_ids.append(sid)
+            self._modify_session_locked(parent_id, _apply, error_label="update parent children")
         
         self._save_extended_session(session)
         return sid
@@ -356,11 +382,15 @@ class HierarchicalSessionStore(DefaultSessionStore):
             metadata=copy.deepcopy(parent.metadata),
         )
         
-        # Update parent's children list
-        parent.children_ids.append(new_id)
-        
         self._save_extended_session(forked)
-        self._save_extended_session(parent)
+
+        def _register_fork(parent: SessionData) -> None:
+            if new_id not in parent.children_ids:
+                parent.children_ids.append(new_id)
+
+        self._modify_session_locked(
+            session_id, _register_fork, error_label="register forked session"
+        )
         
         return new_id
     
@@ -411,18 +441,22 @@ class HierarchicalSessionStore(DefaultSessionStore):
         Returns:
             The snapshot ID
         """
-        # Force reload to get latest messages
-        session = self._load_extended_session(session_id, force_reload=True)
-        
         snapshot = SessionSnapshot(
             session_id=session_id,
-            message_index=len(session.messages) - 1 if session.messages else -1,
+            message_index=-1,  # Set from fresh session inside locked write
             label=label,
             metadata=metadata or {},
         )
-        
-        session.snapshots.append(snapshot)
-        self._save_extended_session(session)
+
+        def _record_snapshot(session: SessionData) -> None:
+            snapshot.message_index = (
+                len(session.messages) - 1 if session.messages else -1
+            )
+            session.snapshots.append(snapshot)
+
+        self._modify_session_locked(
+            session_id, _record_snapshot, error_label="create snapshot"
+        )
         
         return snapshot.id
     
@@ -442,26 +476,30 @@ class HierarchicalSessionStore(DefaultSessionStore):
         Returns:
             True if successful
         """
-        session = self._load_extended_session(session_id)
-        
-        # Find the snapshot
-        snapshot = None
-        for s in session.snapshots:
-            if s.id == snapshot_id:
-                snapshot = s
-                break
-        
-        if snapshot is None:
-            logger.warning(f"Snapshot {snapshot_id} not found")
+        def _apply(session: SessionData) -> None:
+            assert isinstance(session, ExtendedSessionData)
+            
+            # Find the snapshot
+            snapshot = None
+            for s in session.snapshots:
+                if s.id == snapshot_id:
+                    snapshot = s
+                    break
+            
+            if snapshot is None:
+                logger.warning(f"Snapshot {snapshot_id} not found")
+                raise ValueError(f"Snapshot {snapshot_id} not found")
+            
+            # Revert messages
+            if snapshot.message_index >= 0:
+                session.messages = session.messages[:snapshot.message_index + 1]
+            else:
+                session.messages = []
+                
+        try:
+            return self._modify_session_locked(session_id, _apply, error_label="revert to snapshot")
+        except ValueError:
             return False
-        
-        # Revert messages
-        if snapshot.message_index >= 0:
-            session.messages = session.messages[:snapshot.message_index + 1]
-        else:
-            session.messages = []
-        
-        return self._save_extended_session(session)
     
     def revert_to_message(self, session_id: str, message_index: int) -> bool:
         """
@@ -474,27 +512,33 @@ class HierarchicalSessionStore(DefaultSessionStore):
         Returns:
             True if successful
         """
-        # Force reload to get latest messages
-        session = self._load_extended_session(session_id, force_reload=True)
-        
-        if message_index < 0 or message_index >= len(session.messages):
-            logger.warning(f"Invalid message index {message_index}")
+        def _apply(session: SessionData) -> None:
+            assert isinstance(session, ExtendedSessionData)
+            
+            if message_index < 0 or message_index >= len(session.messages):
+                logger.warning(f"Invalid message index {message_index}")
+                raise ValueError(f"Invalid message index {message_index}")
+            
+            session.messages = session.messages[:message_index + 1]
+            
+        try:
+            return self._modify_session_locked(session_id, _apply, error_label="revert to message")
+        except ValueError:
             return False
-        
-        session.messages = session.messages[:message_index + 1]
-        return self._save_extended_session(session)
     
     def share_session(self, session_id: str) -> bool:
         """Mark a session as shared."""
-        session = self._load_extended_session(session_id)
-        session.is_shared = True
-        return self._save_extended_session(session)
+        def _apply(session: SessionData) -> None:
+            assert isinstance(session, ExtendedSessionData)
+            session.is_shared = True
+        return self._modify_session_locked(session_id, _apply, error_label="share session")
     
     def unshare_session(self, session_id: str) -> bool:
         """Mark a session as not shared."""
-        session = self._load_extended_session(session_id)
-        session.is_shared = False
-        return self._save_extended_session(session)
+        def _apply(session: SessionData) -> None:
+            assert isinstance(session, ExtendedSessionData)
+            session.is_shared = False
+        return self._modify_session_locked(session_id, _apply, error_label="unshare session")
     
     def is_shared(self, session_id: str) -> bool:
         """Check if a session is shared."""
@@ -503,9 +547,10 @@ class HierarchicalSessionStore(DefaultSessionStore):
     
     def set_title(self, session_id: str, title: str) -> bool:
         """Set session title."""
-        session = self._load_extended_session(session_id)
-        session.title = title
-        return self._save_extended_session(session)
+        def _apply(session: SessionData) -> None:
+            assert isinstance(session, ExtendedSessionData)
+            session.title = title
+        return self._modify_session_locked(session_id, _apply, error_label="set session title")
     
     async def auto_title(self, session_id: str) -> bool:
         """Generate and set title automatically from first exchange.
@@ -560,12 +605,19 @@ class HierarchicalSessionStore(DefaultSessionStore):
             title = await generate_title_async(user_msg, assistant_msg)
             
             if title and title.strip():
-                # Reload session to avoid overwriting concurrent updates
-                fresh_session = await asyncio.to_thread(self._load_extended_session, session_id)
-                # Only set title if it's still empty
-                if not fresh_session.title or not fresh_session.title.strip():
-                    fresh_session.title = title.strip()
-                    return await asyncio.to_thread(self._save_extended_session, fresh_session)
+                # Use locked read-modify-write to avoid overwriting concurrent updates
+                def _apply(fresh_session: SessionData) -> None:
+                    assert isinstance(fresh_session, ExtendedSessionData)
+                    # Only set title if it's still empty
+                    if not fresh_session.title or not fresh_session.title.strip():
+                        fresh_session.title = title.strip()
+                        
+                return await asyncio.to_thread(
+                    self._modify_session_locked, 
+                    session_id, 
+                    _apply, 
+                    error_label="auto-title session"
+                )
                 
         except Exception as e:
             # Title generation failed - log with context instead of silent failure
@@ -577,7 +629,17 @@ class HierarchicalSessionStore(DefaultSessionStore):
     
     def get_extended_session(self, session_id: str) -> ExtendedSessionData:
         """Get extended session data."""
-        return self._load_extended_session(session_id)
+        return self._read_session_fresh(session_id)
+
+    def invalidate_cache(self, session_id: Optional[str] = None) -> None:
+        """Invalidate base and extended in-memory caches atomically."""
+        with self._lock:
+            if session_id:
+                self._cache.pop(session_id, None)
+                self._extended_cache.pop(session_id, None)
+            else:
+                self._cache.clear()
+                self._extended_cache.clear()
     
     def export_session(self, session_id: str) -> Dict[str, Any]:
         """

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

@@ -15,10 +15,10 @@ import threading
 import time
 
 # fcntl is Unix-only; on Windows, use msvcrt for file locking
-if sys.platform != 'win32':
+try:
     import fcntl
     _HAS_FCNTL = True
-else:
+except ImportError:
     _HAS_FCNTL = False
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -167,11 +167,11 @@ class FileLock:
                     if _HAS_FCNTL:
                         fcntl.flock(self._lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                     else:
-                        # Warn once about degraded locking (should never happen with current platform detection)
+                        # Warn once about degraded locking on non-Windows platforms without fcntl
                         global _WARNED_NO_FCNTL
                         if not _WARNED_NO_FCNTL:
                             logger.warning(
-                                "File locking unavailable on this platform (no fcntl/msvcrt); "
+                                "File locking unavailable on this platform (fcntl not available); "
                                 "concurrent writers may corrupt session files."
                             )
                             _WARNED_NO_FCNTL = True
@@ -265,32 +265,22 @@ class DefaultSessionStore:
     def _load_session(self, session_id: str) -> SessionData:
         """Load session from disk with file locking."""
         filepath = self._get_session_path(session_id)
-        
-        # Check cache first
-        with self._lock:
-            if session_id in self._cache:
-                return self._cache[session_id]
-        
-        # Load from disk
-        if not os.path.exists(filepath):
-            session = SessionData(session_id=session_id)
+
+        # When a session file exists, always reload under FileLock so reads
+        # from another DefaultSessionStore instance (or process) are visible.
+        if os.path.exists(filepath):
+            with FileLock(filepath, self.lock_timeout):
+                session = self._load_session_from_disk(session_id, filepath)
             with self._lock:
                 self._cache[session_id] = session
             return session
-        
-        with FileLock(filepath, self.lock_timeout):
-            try:
-                with open(filepath, "r", encoding="utf-8") as f:
-                    data = json.load(f)
-                session = SessionData.from_dict(data)
-            except (json.JSONDecodeError, IOError) as e:
-                logger.warning(f"Failed to load session {session_id}: {e}")
-                session = SessionData(session_id=session_id)
-        
+
         with self._lock:
+            if session_id in self._cache:
+                return self._cache[session_id]
+            session = SessionData(session_id=session_id)
             self._cache[session_id] = session
-        
-        return session
+            return session
 
     def _load_session_from_disk(self, session_id: str, filepath: str) -> SessionData:
         """Load session JSON from disk (caller must hold FileLock)."""
@@ -540,6 +530,29 @@ class DefaultSessionStore:
             error_label="clear session",
         )
 
+    def set_chat_history(
+        self,
+        session_id: str,
+        messages: List[Dict[str, str]],
+    ) -> bool:
+        """Replace session messages atomically (file-locked read-modify-write)."""
+
+        def _apply(session: SessionData) -> None:
+            session.messages.clear()
+            for msg in messages:
+                session.messages.append(
+                    SessionMessage(
+                        role=msg.get("role", "user"),
+                        content=msg.get("content", ""),
+                        timestamp=msg.get("timestamp", time.time()),
+                        metadata=msg.get("metadata", {}),
+                    )
+                )
+
+        return self._modify_session_locked(
+            session_id, _apply, error_label="set chat history"
+        )
+
     def update_session_metadata(self, session_id: str, **fields: Any) -> bool:
         """Merge run stats / metadata fields into a persisted session."""
         if not fields:

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

@@ -0,0 +1,584 @@
+"""
+Tool Search - Progressive disclosure layer for MCP and non-core plugin tools.
+
+Implements Hermes-parity bridge tools with lazy-loaded implementation:
+- tool_search(query) → matches
+- tool_describe(name) → full schema  
+- tool_call(name, args) → unwrap → execute_tool(real_name)
+
+Architecture invariants (ported from Hermes):
+1. Core tools never defer - PRAISONAI_CORE_TOOLS
+2. Unknown tools stay visible - never silently dropped
+3. Stateless catalog - rebuilt every assembly from current tool-defs
+4. skip_assembly for bridge dispatch - handlers read pre-assembly deferrable set
+5. Session scoping - catalog = deferrable subset of agent's own tool list
+6. Unwrap before trace/stream - hooks, approval, AG-UI show real tool name
+7. Auto threshold - deferrable schema tokens ≥ threshold_pct * context_length
+8. BM25 inlined - no new dependency; substring fallback on tool name
+"""
+
+import re
+import math
+from typing import Dict, List, Optional, Union, Protocol, Set, FrozenSet, Tuple, Any
+from dataclasses import dataclass, field
+from collections import defaultdict
+
+# Type definitions matching Hermes patterns
+ToolDef = Dict[str, Any]  # OpenAI function schema
+ToolDefList = List[ToolDef]
+
+# Core tools that never defer (mirrors Hermes _HERMES_CORE_TOOLS)
+PRAISONAI_CORE_TOOLS: FrozenSet[str] = frozenset({
+    # File operations
+    "read_file", "write_file", "list_files", "get_file_info", 
+    "copy_file", "move_file", "delete_file",
+    
+    # Shell operations  
+    "execute_command", "list_processes", "kill_process", "get_system_info",
+    
+    # Web operations
+    "search_web", "web_search", "internet_search", "web_crawl", "crawl_web",
+    
+    # Schedule operations (agent-centric scheduling)
+    "schedule_add", "schedule_list", "schedule_remove",
+    
+    # Memory operations (active memory)
+    "store_memory", "search_memory",
+    
+    # Clarify tool (human-in-the-loop clarification)
+    "clarify",
+})
+
+@dataclass
+class ToolSearchConfig:
+    """Configuration for Tool Search feature."""
+    enabled: Union[bool, str] = "auto"  # auto | on | off | True | False
+    threshold_pct: float = 10.0  # Percentage of context window for deferral threshold
+    search_default_limit: int = 5  # Default number of search results
+    max_search_limit: int = 20  # Maximum search results allowed
+    core_tools: Optional[FrozenSet[str]] = None  # Override core tools set
+    
+    @classmethod
+    def from_raw(cls, raw_config: Any) -> "ToolSearchConfig":
+        """Create config from various input formats (bool, dict, ToolSearchConfig)."""
+        if isinstance(raw_config, cls):
+            return raw_config
+        elif isinstance(raw_config, bool):
+            return cls(enabled="on" if raw_config else "off")
+        elif isinstance(raw_config, str):
+            if raw_config.lower() in ("true", "on", "yes", "1"):
+                return cls(enabled="on")
+            elif raw_config.lower() in ("false", "off", "no", "0"):
+                return cls(enabled="off") 
+            elif raw_config.lower() == "auto":
+                return cls(enabled="auto")
+            else:
+                raise ValueError(f"Invalid tool_search string value: {raw_config}")
+        elif isinstance(raw_config, dict):
+            return cls(**raw_config)
+        else:
+            raise TypeError(f"Invalid tool_search type: {type(raw_config)}")
+
+class ToolSearchProtocol(Protocol):
+    """Protocol for tool search implementations."""
+    
+    def search_catalog(self, query: str, limit: int = 5) -> List[Dict[str, str]]:
+        """Search for tools matching the query."""
+        ...
+        
+    def describe_tool(self, tool_name: str) -> Optional[Dict[str, Any]]:
+        """Get full schema for a specific tool."""
+        ...
+        
+    def resolve_underlying_call(self, tool_name: str, tool_args: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]:
+        """Unwrap bridge call to real tool name and args."""
+        ...
+
+def classify_tools(tool_defs: ToolDefList, config: ToolSearchConfig) -> Tuple[ToolDefList, ToolDefList]:
+    """
+    Classify tools into core (never defer) vs deferrable.
+    
+    Args:
+        tool_defs: List of OpenAI function schemas
+        config: Tool search configuration
+        
+    Returns:
+        Tuple of (core_tools, deferrable_tools)
+    """
+    core_tools = []
+    deferrable_tools = []
+    
+    # Use configured core tools or default
+    core_tool_names = config.core_tools or PRAISONAI_CORE_TOOLS
+    
+    for tool_def in tool_defs:
+        tool_name = tool_def.get("function", {}).get("name", "")
+        
+        # Check if it's a core tool
+        if tool_name in core_tool_names:
+            core_tools.append(tool_def)
+            continue
+            
+        # Check for deferrable marker from MCP or custom tools
+        if _is_tool_deferrable(tool_def):
+            deferrable_tools.append(tool_def)
+        else:
+            # Unknown tools stay visible (invariant #2)
+            core_tools.append(tool_def)
+    
+    return core_tools, deferrable_tools
+
+def _is_tool_deferrable(tool_def: ToolDef) -> bool:
+    """Check if a tool should be deferred based on markers."""
+    # Check for MCP deferrable marker
+    if tool_def.get("__praisonai_deferrable__", False):
+        return True
+        
+    # Check function metadata for deferrable flag
+    function_def = tool_def.get("function", {})
+    if function_def.get("deferrable", False):
+        return True
+        
+    # Check tool name patterns for MCP tools (prefix only, to avoid false-positives)
+    tool_name = function_def.get("name", "")
+    if tool_name.startswith("mcp_"):
+        return True
+        
+    return False
+
+def estimate_tool_schema_tokens(tool_defs: ToolDefList) -> int:
+    """
+    Estimate token count for tool schemas.
+    
+    Rough approximation: 1 token ≈ 4 characters for English text.
+    Function schemas are structured JSON so may be more dense.
+    """
+    if not tool_defs:
+        return 0
+        
+    # Convert to JSON and estimate
+    import json
+    total_chars = 0
+    for tool_def in tool_defs:
+        try:
+            json_str = json.dumps(tool_def, separators=(',', ':'))
+            total_chars += len(json_str)
+        except (TypeError, ValueError):
+            # Fallback estimation
+            total_chars += len(str(tool_def))
+    
+    # Conservative estimate: 1 token per 3.5 characters for JSON
+    return int(total_chars / 3.5)
+
+def should_defer_tools(
+    deferrable_tools: ToolDefList, 
+    config: ToolSearchConfig,
+    context_length: Optional[int] = None
+) -> bool:
+    """
+    Determine if tools should be deferred based on auto threshold.
+    
+    Args:
+        deferrable_tools: List of deferrable tool schemas
+        config: Tool search configuration  
+        context_length: Model context window size
+        
+    Returns:
+        True if tools should be deferred
+    """
+    if config.enabled == "off" or config.enabled is False:
+        return False
+    elif config.enabled == "on" or config.enabled is True:
+        return True
+    elif config.enabled == "auto":
+        # Auto mode: check threshold
+        if not deferrable_tools:
+            return False
+            
+        deferrable_tokens = estimate_tool_schema_tokens(deferrable_tools)
+        
+        # Use fallback if context length unknown
+        context_limit = context_length or 20000
+        threshold_tokens = int(context_limit * (config.threshold_pct / 100.0))
+        
+        return deferrable_tokens >= threshold_tokens
+    else:
+        return False
+
+class BM25ToolSearcher:
+    """
+    Simple BM25 implementation for tool search.
+    No external dependencies - inlined from Hermes approach.
+    """
+    
+    def __init__(self, tool_catalog: List[Dict[str, str]]):
+        """
+        Initialize with tool catalog.
+        
+        Args:
+            tool_catalog: List of dicts with 'name', 'description' keys
+        """
+        self.catalog = tool_catalog
+        self.term_frequencies = []
+        self.doc_frequencies = defaultdict(int)
+        self.total_docs = len(tool_catalog)
+        
+        # Preprocess documents
+        self._build_index()
+    
+    def _tokenize(self, text: str) -> List[str]:
+        """Simple tokenization - lowercase, alphanumeric + underscores."""
+        if not text:
+            return []
+        return re.findall(r'\b[a-zA-Z0-9_]+\b', text.lower())
+    
+    def _build_index(self):
+        """Build BM25 index from catalog."""
+        total_length = 0
+        # Calculate term frequencies for each document
+        for item in self.catalog:
+            doc_text = f"{item['name']} {item['description']}"
+            tokens = self._tokenize(doc_text)
+            
+            # Term frequency for this doc
+            tf = defaultdict(int)
+            for token in tokens:
+                tf[token] += 1
+                
+            self.term_frequencies.append(dict(tf))
+            total_length += len(tokens)
+            
+            # Document frequency (how many docs contain each term)
+            unique_tokens = set(tokens)
+            for token in unique_tokens:
+                self.doc_frequencies[token] += 1
+        
+        # Cache average document length to avoid O(n) recomputation in hot path
+        self._cached_avg_doc_length = total_length / len(self.catalog) if self.catalog else 0.0
+    
+    def search(self, query: str, limit: int = 5) -> List[Dict[str, str]]:
+        """
+        Search using BM25 scoring.
+        
+        Args:
+            query: Search query string
+            limit: Maximum results to return
+            
+        Returns:
+            List of matching tool items, sorted by relevance
+        """
+        query_tokens = self._tokenize(query)
+        if not query_tokens:
+            return []
+        
+        scores = []
+        k1, b = 1.5, 0.75  # BM25 parameters
+        
+        for item, tf in zip(self.catalog, self.term_frequencies, strict=True):
+            score = 0.0
+            doc_length = sum(tf.values())
+            
+            for token in query_tokens:
+                if token in tf:
+                    # BM25 formula
+                    idf = math.log((self.total_docs - self.doc_frequencies[token] + 0.5) / 
+                                 (self.doc_frequencies[token] + 0.5))
+                    term_freq = tf[token]
+                    score += idf * (term_freq * (k1 + 1)) / (
+                        term_freq + k1 * (1 - b + b * (doc_length / self._cached_avg_doc_length))
+                    )
+            
+            if score > 0:
+                scores.append((score, item))
+        
+        # Sort by score descending and return top results
+        scores.sort(key=lambda x: x[0], reverse=True)
+        return [item for _, item in scores[:limit]]
+    
+
+def search_catalog(
+    deferrable_tools: ToolDefList, 
+    query: str, 
+    limit: int = 5
+) -> List[Dict[str, str]]:
+    """
+    Search deferrable tools catalog using BM25.
+    
+    Args:
+        deferrable_tools: List of deferrable tool schemas
+        query: Search query
+        limit: Maximum results
+        
+    Returns:
+        List of matching tools with name/description
+    """
+    if not deferrable_tools or not query.strip():
+        return []
+    
+    # Build catalog for searching
+    catalog = []
+    for tool_def in deferrable_tools:
+        function_def = tool_def.get("function", {})
+        name = function_def.get("name", "")
+        description = function_def.get("description", "")
+        
+        if name:  # Only include tools with names
+            catalog.append({
+                "name": name,
+                "description": description
+            })
+    
+    if not catalog:
+        return []
+    
+    # Use BM25 search if query is substantial, otherwise substring fallback
+    if len(query.strip()) >= 3:
+        searcher = BM25ToolSearcher(catalog)
+        results = searcher.search(query, limit)
+        
+        # If BM25 finds results, use them
+        if results:
+            return results
+    
+    # Fallback: simple substring matching on tool names
+    query_lower = query.lower()
+    matches = []
+    for item in catalog:
+        if query_lower in item["name"].lower() or query_lower in item["description"].lower():
+            matches.append(item)
+            if len(matches) >= limit:
+                break
+                
+    return matches
+
+def bridge_tool_schemas() -> ToolDefList:
+    """
+    Generate the three bridge tool schemas.
+    
+    Returns:
+        List of OpenAI function schemas for bridge tools
+    """
+    return [
+        {
+            "type": "function",
+            "function": {
+                "name": "tool_search",
+                "description": "Search for available tools by name or functionality. Use this to discover what tools are available before using them.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "query": {
+                            "type": "string",
+                            "description": "Search query describing the tool functionality you need"
+                        },
+                        "limit": {
+                            "type": "integer",
+                            "description": "Maximum number of results to return",
+                            "default": 5,
+                            "minimum": 1,
+                            "maximum": 20
+                        }
+                    },
+                    "required": ["query"]
+                }
+            }
+        },
+        {
+            "type": "function", 
+            "function": {
+                "name": "tool_describe",
+                "description": "Get the full schema and documentation for a specific tool. Use this after tool_search to understand how to use a tool.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "tool_name": {
+                            "type": "string",
+                            "description": "Exact name of the tool to describe"
+                        }
+                    },
+                    "required": ["tool_name"]
+                }
+            }
+        },
+        {
+            "type": "function",
+            "function": {
+                "name": "tool_call", 
+                "description": "Execute a tool with the given arguments. Use this after tool_describe to actually run the tool.",
+                "parameters": {
+                    "type": "object",
+                    "properties": {
+                        "tool_name": {
+                            "type": "string",
+                            "description": "Exact name of the tool to execute"
+                        },
+                        "tool_args": {
+                            "type": "object", 
+                            "description": "Arguments to pass to the tool",
+                            "additionalProperties": True
+                        }
+                    },
+                    "required": ["tool_name", "tool_args"]
+                }
+            }
+        }
+    ]
+
+def assemble_tool_defs(
+    tool_defs: ToolDefList,
+    config: ToolSearchConfig, 
+    context_length: Optional[int] = None
+) -> Tuple[ToolDefList, Dict[str, Any]]:
+    """
+    Main assembly function - decide whether to use bridge mode or pass-through.
+    
+    Args:
+        tool_defs: Original list of tool schemas
+        config: Tool search configuration
+        context_length: Model context window size
+        
+    Returns:
+        Tuple of (assembled_tool_defs, metadata)
+        metadata contains: {"bridge_mode": bool, "deferred_count": int, "catalog": List}
+    """
+    # Always return original tools if disabled
+    if config.enabled == "off" or config.enabled is False:
+        return tool_defs, {"bridge_mode": False, "deferred_count": 0, "catalog": []}
+    
+    # Classify tools
+    core_tools, deferrable_tools = classify_tools(tool_defs, config)
+    
+    # Check if we should defer
+    should_defer = should_defer_tools(deferrable_tools, config, context_length)
+    
+    if not should_defer or not deferrable_tools:
+        # Pass-through mode - return original tools
+        return tool_defs, {"bridge_mode": False, "deferred_count": 0, "catalog": []}
+    
+    # Bridge mode - replace deferrable tools with bridge schemas
+    bridge_schemas = bridge_tool_schemas()
+    assembled_tools = core_tools + bridge_schemas
+    
+    # Build catalog for search
+    catalog = []
+    for tool_def in deferrable_tools:
+        function_def = tool_def.get("function", {})
+        name = function_def.get("name", "")
+        description = function_def.get("description", "")
+        if name:
+            catalog.append({"name": name, "description": description})
+    
+    metadata = {
+        "bridge_mode": True,
+        "deferred_count": len(deferrable_tools), 
+        "catalog": catalog,
+        "deferrable_tools": deferrable_tools  # Keep full schemas for describe/call
+    }
+    
+    return assembled_tools, metadata
+
+def dispatch_tool_search(
+    query: str, 
+    limit: Optional[int],
+    deferrable_tools: ToolDefList,
+    config: ToolSearchConfig
+) -> Dict[str, Any]:
+    """
+    Handle tool_search bridge call.
+    
+    Returns:
+        JSON response with search results
+    """
+    # Validate and clamp limit
+    search_limit = limit or config.search_default_limit
+    search_limit = min(search_limit, config.max_search_limit)
+    search_limit = max(search_limit, 1)
+    
+    # Perform search
+    results = search_catalog(deferrable_tools, query, search_limit)
+    
+    return {
+        "query": query,
+        "results": results,
+        "total_available": len(deferrable_tools)
+    }
+
+def dispatch_tool_describe(
+    tool_name: str,
+    deferrable_tools: ToolDefList
+) -> Dict[str, Any]:
+    """
+    Handle tool_describe bridge call.
+    
+    Returns:
+        JSON response with full tool schema or error
+    """
+    # Find the tool
+    for tool_def in deferrable_tools:
+        function_def = tool_def.get("function", {})
+        if function_def.get("name") == tool_name:
+            return {
+                "tool_name": tool_name,
+                "schema": tool_def,
+                "found": True
+            }
+    
+    return {
+        "tool_name": tool_name,
+        "error": f"Tool '{tool_name}' not found in available tools",
+        "found": False
+    }
+
+def resolve_underlying_call(tool_name: str, tool_args: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]:
+    """
+    Unwrap tool_call bridge to get real tool name and args.
+    
+    Args:
+        tool_name: Should be "tool_call"
+        tool_args: Dict with "tool_name" and "tool_args" keys
+        
+    Returns:
+        Tuple of (real_tool_name, real_args)
+    """
+    if tool_name != "tool_call":
+        # Not a bridge call, return as-is
+        return tool_name, tool_args
+    
+    # Validate tool_args is a dict
+    if not isinstance(tool_args, dict):
+        raise TypeError(
+            f"tool_call expects a dictionary for tool_args, got {type(tool_args).__name__}. "
+            "Ensure the LLM output is properly formatted."
+        )
+    
+    # Extract real tool call from bridge args
+    real_tool_name = tool_args.get("tool_name", "")
+    real_args = tool_args.get("tool_args", {})
+    
+    if not real_tool_name:
+        raise ValueError("tool_call requires 'tool_name' parameter")
+    
+    return real_tool_name, real_args
+
+def scoped_deferrable_names(agent_tool_list: List[str], deferrable_tools: ToolDefList) -> Set[str]:
+    """
+    Get names of deferrable tools that are in the agent's enabled tool list.
+    Implements session scoping (invariant #5).
+    
+    Args:
+        agent_tool_list: List of tool names available to this agent
+        deferrable_tools: Full list of deferrable tool schemas
+        
+    Returns:
+        Set of tool names that are both deferrable and enabled for this agent
+    """
+    agent_tool_set = set(agent_tool_list)
+    deferrable_names = set()
+    
+    for tool_def in deferrable_tools:
+        function_def = tool_def.get("function", {})
+        tool_name = function_def.get("name", "")
+        if tool_name and tool_name in agent_tool_set:
+            deferrable_names.add(tool_name)
+    
+    return deferrable_names

src/praisonai-agents/pyproject.toml

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

src/praisonai-agents/tests/managed/test_managed_persistence.py

@@ -349,6 +349,20 @@ class TestDbSessionAdapter:
         assert meta["agent_id"] == "abc"
         assert meta["total_input_tokens"] == 42
 
+    def test_update_session_metadata_accepts_keyword_fields(self):
+        """update_session_metadata matches DefaultSessionStore **fields API."""
+        from praisonai.integrations.db_session_adapter import DbSessionAdapter
+
+        mock_db = MagicMock()
+        adapter = DbSessionAdapter(mock_db)
+        adapter.update_session_metadata(
+            "sess1", model="gpt-4o-mini", total_tokens=99, cost=0.01
+        )
+        meta = adapter.get_metadata("sess1")
+        assert meta["model"] == "gpt-4o-mini"
+        assert meta["total_tokens"] == 99
+        assert meta["cost"] == 0.01
+
     def test_clear_session_purges_db_history(self):
         """clear_session must not leave stale messages reloadable from the DB."""
         from praisonai.integrations.db_session_adapter import DbSessionAdapter

src/praisonai-agents/tests/unit/agent/test_chat_sync_stream_defaults.py

@@ -0,0 +1,48 @@
+import ast
+from pathlib import Path
+
+
+def _chat_mixin_path() -> Path:
+    for candidate in Path(__file__).resolve().parents:
+        chat_mixin = candidate / "praisonaiagents" / "agent" / "chat_mixin.py"
+        if chat_mixin.exists():
+            return chat_mixin
+    raise FileNotFoundError("Could not locate praisonaiagents/agent/chat_mixin.py")
+
+
+def get_chat_mixin_stream_parameter_default(function_name: str):
+    chat_mixin = _chat_mixin_path()
+    tree = ast.parse(chat_mixin.read_text(encoding="utf-8"))
+
+    for node in tree.body:
+        if isinstance(node, ast.ClassDef) and node.name == "ChatMixin":
+            for item in node.body:
+                if isinstance(item, ast.FunctionDef) and item.name == function_name:
+                    args = item.args.args
+                    defaults = item.args.defaults
+                    first_default_index = len(args) - len(defaults)
+                    for idx, arg in enumerate(args):
+                        if arg.arg == "stream":
+                            default_idx = idx - first_default_index
+                            if default_idx < 0:
+                                raise AssertionError(
+                                    f"Function '{function_name}' has no default for stream"
+                                )
+                            default = defaults[default_idx]
+                            if isinstance(default, ast.Constant):
+                                return default.value
+                            raise AssertionError(
+                                f"Function '{function_name}' stream default is non-constant"
+                            )
+    raise AssertionError(f"Function '{function_name}' or stream parameter not found")
+
+
+def test_chat_completion_stream_default_is_none():
+    assert get_chat_mixin_stream_parameter_default("_chat_completion") is None
+
+
+def test_execute_unified_chat_completion_stream_default_is_none():
+    assert (
+        get_chat_mixin_stream_parameter_default("_execute_unified_chat_completion")
+        is None
+    )

src/praisonai-agents/tests/unit/mcp/test_mcp_tool_runner_concurrency.py

@@ -0,0 +1,75 @@
+"""Regression tests for MCPToolRunner concurrent call routing."""
+
+import queue
+import threading
+import time
+from unittest.mock import Mock, patch
+
+import pytest
+
+
+class TestMCPToolRunnerConcurrency:
+    def test_concurrent_calls_receive_matching_results(self):
+        from praisonaiagents.mcp.mcp import MCPToolRunner
+
+        with patch.object(MCPToolRunner, "start", lambda self: None):
+            runner = MCPToolRunner(server_params=Mock(), timeout=5)
+        runner.initialized.set()
+
+        results = {}
+        barrier = threading.Barrier(2)
+
+        def slow_worker():
+            while True:
+                item = runner.queue.get()
+                if item is None:
+                    break
+                response_queue, tool_name, _arguments = item
+                if tool_name == "slow_tool":
+                    time.sleep(0.05)
+                    response_queue.put((True, "slow-result"))
+                else:
+                    response_queue.put((True, "fast-result"))
+
+        worker = threading.Thread(target=slow_worker, daemon=True)
+        worker.start()
+
+        def call_tool(name):
+            barrier.wait()
+            results[name] = runner.call_tool(name, {})
+
+        threads = [
+            threading.Thread(target=call_tool, args=("slow_tool",)),
+            threading.Thread(target=call_tool, args=("fast_tool",)),
+        ]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join(timeout=5)
+
+        runner.queue.put(None)
+        worker.join(timeout=2)
+
+        assert results["slow_tool"] == "slow-result"
+        assert results["fast_tool"] == "fast-result"
+
+    def test_call_tool_times_out_when_worker_stalls(self):
+        from praisonaiagents.mcp.mcp import MCPToolRunner
+
+        with patch.object(MCPToolRunner, "start", lambda self: None):
+            runner = MCPToolRunner(server_params=Mock(), timeout=1)
+        runner.initialized.set()
+
+        result = runner.call_tool("stalled_tool", {})
+        assert "timed out" in result.lower()
+
+    def test_init_error_is_not_returned_to_unrelated_callers(self):
+        from praisonaiagents.mcp.mcp import MCPToolRunner
+
+        with patch.object(MCPToolRunner, "start", lambda self: None):
+            runner = MCPToolRunner(server_params=Mock(), timeout=5)
+        runner.initialized.set()
+        runner._init_error = "MCP initialization error: boom"
+
+        result = runner.call_tool("any_tool", {})
+        assert result == "Error: MCP initialization error: boom"

src/praisonai-agents/tests/unit/session/test_hierarchy.py

@@ -117,6 +117,44 @@ class TestHierarchicalSessionStore:
             assert len(history) == 2
             assert history[1]["content"] == "second"
 
+    def test_get_extended_session_sees_writes_from_other_store(self):
+        """Extended reads must reload from disk, not stale _extended_cache."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            writer = HierarchicalSessionStore(session_dir=tmpdir)
+            reader = HierarchicalSessionStore(session_dir=tmpdir)
+
+            writer.add_user_message("session-1", "first")
+            reader._load_extended_session("session-1")
+            writer.add_user_message("session-1", "second")
+
+            session = reader.get_extended_session("session-1")
+            assert len(session.messages) == 2
+            assert session.messages[1].content == "second"
+
+    def test_stale_cache_write_preserves_concurrent_updates(self):
+        """Metadata writes must not truncate messages written by other processes."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            writer = HierarchicalSessionStore(session_dir=tmpdir)
+            reader = HierarchicalSessionStore(session_dir=tmpdir)
+            
+            # Create session and warm reader's cache
+            writer.create_session("session-1", title="Original")
+            reader.get_extended_session("session-1")  # Warms cache
+            
+            # Writer adds messages, reader has stale cache
+            writer.add_user_message("session-1", "first message")
+            writer.add_assistant_message("session-1", "first response")
+            
+            # Reader performs metadata-only write with stale cache
+            reader.set_title("session-1", "Updated Title")
+            
+            # Verify messages are preserved
+            session = writer.get_extended_session("session-1")
+            assert session.title == "Updated Title"
+            assert len(session.messages) == 2
+            assert session.messages[0].content == "first message"
+            assert session.messages[1].content == "first response"
+
     def test_update_session_metadata_preserves_extended_fields(self):
         """Metadata updates must not strip parent_id, snapshots, etc."""
         session_id = self.store.create_session(title="Parent")
@@ -324,6 +362,94 @@ class TestHierarchicalSessionStore:
         new_id = self.store.import_session(exported, new_session_id=custom_id)
         
         assert new_id == custom_id
+    
+    def test_set_title_does_not_drop_messages_after_external_write(self):
+        """
+        Regression test for the stale cache bug.
+        
+        Reproduces the scenario where:
+        1. Process A loads session (warms cache)
+        2. Process B writes new messages
+        3. Process A calls set_title() → should NOT drop Process B's messages
+        """
+        import json
+        
+        # Create session with initial messages
+        session_id = self.store.create_session(title="Test Session")
+        self.store.add_message(session_id, "user", "Message 1")
+        self.store.add_message(session_id, "assistant", "Response 1")
+        
+        # Process A: Load session (warms cache)
+        session_a = self.store.get_extended_session(session_id)
+        assert len(session_a.messages) == 2
+        
+        # Process B: Simulate external write by directly modifying file
+        # This mimics another process/store instance writing to the same session
+        filepath = self.store._get_session_path(session_id)
+        with open(filepath, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        
+        # Add messages from "Process B"
+        data["messages"].extend([
+            {"role": "user", "content": "Message 2", "timestamp": time.time(), "metadata": {}},
+            {"role": "assistant", "content": "Response 2", "timestamp": time.time(), "metadata": {}}
+        ])
+        data["updated_at"] = time.time()
+        
+        # Write the updated data (simulating external process write)
+        with open(filepath, "w", encoding="utf-8") as f:
+            json.dump(data, f, indent=2)
+        
+        # Brief sleep to ensure file mtime is different
+        time.sleep(0.1)
+        
+        # Process A: Call set_title() - this should detect the external write
+        # and reload fresh data instead of using stale cache
+        result = self.store.set_title(session_id, "Updated Title")
+        assert result is True
+        
+        # Verify no messages were lost - should have all 4 messages
+        final_session = self.store.get_extended_session(session_id)
+        assert len(final_session.messages) == 4, f"Expected 4 messages, got {len(final_session.messages)}"
+        assert final_session.title == "Updated Title"
+        assert final_session.messages[0].content == "Message 1"
+        assert final_session.messages[1].content == "Response 1"
+        assert final_session.messages[2].content == "Message 2"
+        assert final_session.messages[3].content == "Response 2"
+    
+    def test_cache_performance_with_unchanged_files(self):
+        """
+        Test that performance optimization works - reads from cache when file hasn't changed.
+        """
+        session_id = self.store.create_session(title="Cache Test")
+        self.store.add_message(session_id, "user", "Test message")
+        
+        # First read - loads from disk and caches
+        session1 = self.store.get_extended_session(session_id)
+        assert len(session1.messages) == 1
+        
+        # Second read should use cache (file hasn't changed)
+        # We can't easily test this directly, but we can verify the cache is valid
+        assert self.store._is_cache_valid(session_id) is True
+        
+        session2 = self.store.get_extended_session(session_id)
+        assert len(session2.messages) == 1
+        assert session2 is session1  # Should be same cached object
+    
+    def test_force_reload_bypasses_cache(self):
+        """Test that force_reload=True always loads from disk."""
+        session_id = self.store.create_session(title="Force Reload Test")
+        self.store.add_message(session_id, "user", "Message 1")
+        
+        # Load and cache
+        session1 = self.store._load_extended_session(session_id, force_reload=False)
+        
+        # Force reload should bypass cache
+        session2 = self.store._load_extended_session(session_id, force_reload=True)
+        
+        # Both should have same data but force_reload ensures fresh read
+        assert len(session1.messages) == len(session2.messages)
+        assert session1.session_id == session2.session_id
 
 
 class TestGlobalHierarchicalStore:

src/praisonai-agents/tests/unit/session/test_session_store.py

@@ -9,8 +9,11 @@ import os
 import tempfile
 import threading
 import time
+import builtins
+import importlib
 import pytest
 from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import patch
 
 from praisonaiagents.session.store import (
     DefaultSessionStore,
@@ -160,6 +163,26 @@ class TestFileLock:
                         if start_idx < other_start < end_idx:
                             pytest.fail(f"Lock interleaving detected: {results}")
 
+    def test_import_without_fcntl(self):
+        """Test module import succeeds when fcntl is unavailable."""
+        import praisonaiagents.session.store as store_module
+        import importlib.util
+        original_import = builtins.__import__
+
+        def import_without_fcntl(name, *args, **kwargs):
+            if name == "fcntl":
+                raise ImportError("fcntl unavailable")
+            return original_import(name, *args, **kwargs)
+
+        try:
+            with patch("builtins.__import__", side_effect=import_without_fcntl):
+                reloaded_module = importlib.reload(store_module)
+                assert reloaded_module._HAS_FCNTL is False
+        finally:
+            restored_module = importlib.reload(store_module)
+        
+        assert restored_module._HAS_FCNTL is (importlib.util.find_spec("fcntl") is not None)
+
 
 class TestDefaultSessionStore:
     """Tests for DefaultSessionStore."""
@@ -327,6 +350,21 @@ class TestDefaultSessionStore:
             assert session.agent_name == "TestBot"
             assert session.user_id == "u-1"
 
+    def test_set_chat_history_replaces_under_lock(self, temp_store):
+        """History replace is a single locked write (not clear + N adds)."""
+        temp_store.add_user_message("session-1", "old")
+        assert temp_store.set_chat_history(
+            "session-1",
+            [
+                {"role": "user", "content": "new"},
+                {"role": "assistant", "content": "reply"},
+            ],
+        )
+        assert temp_store.get_chat_history("session-1") == [
+            {"role": "user", "content": "new"},
+            {"role": "assistant", "content": "reply"},
+        ]
+
     def test_get_chat_history_sees_writes_from_other_store(self, temp_store):
         """Reads must reload from disk, not a stale in-memory cache."""
         with tempfile.TemporaryDirectory() as tmpdir:
@@ -336,10 +374,10 @@ class TestDefaultSessionStore:
             writer.add_user_message("session-1", "first")
             reader._load_session("session-1")
             writer.add_user_message("session-1", "second")
-
             history = reader.get_chat_history("session-1")
             assert len(history) == 2
             assert history[1]["content"] == "second"
+            assert history[1]["content"] == "second"
 
     def test_clear_session_preserves_new_messages(self, temp_store):
         """Clear must reload from disk so concurrent adds are not lost."""
@@ -421,12 +459,26 @@ class TestDefaultSessionStore:
         
         # Reads always reload from disk (no stale cache)
         history = temp_store.get_chat_history("session-1")
-        assert len(history) == 2  # Fresh from disk
-        
-        # invalidate_cache still works but is now essentially a no-op for reads
+        assert len(history) == 2
+
+        # invalidate_cache still clears in-memory state when a file is missing
         temp_store.invalidate_cache("session-1")
         history = temp_store.get_chat_history("session-1")
         assert len(history) == 2
+
+    def test_get_chat_history_sees_other_store_instance(self):
+        """Another store on the same session_dir must see new messages without invalidate_cache."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            writer = DefaultSessionStore(session_dir=tmpdir)
+            reader = DefaultSessionStore(session_dir=tmpdir)
+
+            writer.add_user_message("session-1", "first")
+            reader._load_session("session-1")
+            writer.add_user_message("session-1", "second")
+
+            history = reader.get_chat_history("session-1")
+            assert len(history) == 2
+            assert history[1]["content"] == "second"
     
     def test_list_sessions_with_none_updated_at(self, temp_store):
         """Test list_sessions handles None updated_at values without crashing.

src/praisonai-agents/tests/unit/test_mcp_loader.py

@@ -0,0 +1,125 @@
+"""
+Unit tests for MCP loader functionality.
+
+Tests the smoke functionality for the new MCP client lifecycle
+features implemented in this issue.
+"""
+
+import pytest
+from unittest.mock import Mock, MagicMock
+from praisonaiagents.memory.mcp_config import MCPConfig
+
+
+def test_mcp_config_to_mcp_instance_import_fix():
+    """Test that MCPConfig.to_mcp_instance() imports MCP correctly (A1)."""
+    # Create a minimal config
+    config = MCPConfig(
+        name="test",
+        command="echo",
+        args=["hello"],
+        enabled=True
+    )
+    
+    # This should not raise ImportError anymore
+    # Note: It might return None if MCP package is not installed, but the import should work
+    try:
+        result = config.to_mcp_instance()
+        # If we get here without ImportError from wrong path, the fix worked
+        assert True
+    except ImportError as e:
+        if "praisonaiagents.tools.mcp" in str(e):
+            pytest.fail("Import still using wrong path - A1 fix failed")
+        else:
+            # This is expected if mcp package is not installed
+            assert "praisonaiagents[mcp]" in str(e)
+
+
+def test_load_mcp_tools_import():
+    """Test that load_mcp_tools can be imported (B2)."""
+    from praisonaiagents.mcp import load_mcp_tools
+    assert callable(load_mcp_tools)
+
+
+def test_mcp_client_protocol_import():
+    """Test that MCPClientProtocol can be imported (B1)."""
+    from praisonaiagents.mcp import MCPClientProtocol
+    # Should be a protocol/type
+    assert MCPClientProtocol is not None
+
+
+def test_mcp_filter_parameters():
+    """Test that MCP class accepts filter parameters (B3)."""
+    try:
+        from praisonaiagents.mcp import MCP
+        # This should not raise TypeError for unknown parameters
+        # Note: May raise ImportError if mcp package not installed
+        mcp = MCP("echo hello", allowed_tools=["test"], disabled_tools=["bad"])
+        assert mcp.allowed_tools == ["test"]
+        assert mcp.disabled_tools == ["bad"]
+    except ImportError as e:
+        if "praisonaiagents[mcp]" in str(e):
+            # Expected if MCP package not installed
+            pytest.skip("MCP package not installed")
+        else:
+            raise
+
+
+def test_filter_tool_functions_exist():
+    """Test that filter functions exist and work (B3)."""
+    from praisonaiagents.mcp.mcp_utils import filter_tools_by_allowlist, filter_disabled_tools
+    
+    tools = [
+        {"name": "tool1", "description": "First tool"},
+        {"name": "tool2", "description": "Second tool"},
+        {"name": "tool3", "description": "Third tool"},
+    ]
+    
+    # Test allowlist filtering
+    filtered = filter_tools_by_allowlist(tools, ["tool1", "tool3"])
+    assert len(filtered) == 2
+    assert filtered[0]["name"] == "tool1"
+    assert filtered[1]["name"] == "tool3"
+    
+    # Test disabled filtering
+    filtered = filter_disabled_tools(tools, ["tool2"])
+    assert len(filtered) == 2
+    assert all(t["name"] != "tool2" for t in filtered)
+    
+    # Test precedence: include wins over exclude
+    # This is handled at MCP class level, but filters should work independently
+
+
+def test_prefix_collision_handling():
+    """Test tool name prefix handling (B4)."""
+    # The prefix logic is in the loader
+    from praisonaiagents.mcp.loader import load_mcp_tools
+    
+    # Test that sanitization logic exists (basic check)
+    # Full test would require mock MCP configs
+    assert callable(load_mcp_tools)
+
+
+def test_safe_env_build():
+    """Test safe environment building (B5)."""
+    try:
+        from praisonaiagents.mcp import MCP
+        
+        # Create MCP instance to test env building
+        # This tests that _build_safe_env method exists
+        mcp = MCP("echo test")
+        assert hasattr(mcp, '_build_safe_env')
+        
+        # Test safe env function
+        safe_env = mcp._build_safe_env({"CUSTOM_VAR": "test"})
+        assert isinstance(safe_env, dict)
+        assert "CUSTOM_VAR" in safe_env
+        assert safe_env["CUSTOM_VAR"] == "test"
+        # Should have safe baseline
+        assert "PATH" in safe_env
+        assert "HOME" in safe_env
+        
+    except ImportError as e:
+        if "praisonaiagents[mcp]" in str(e):
+            pytest.skip("MCP package not installed") 
+        else:
+            raise

src/praisonai-agents/tests/unit/tools/test_tool_search.py

@@ -0,0 +1,394 @@
+"""
+Unit tests for Tool Search functionality.
+
+Tests core tool search components including:
+- ToolSearchConfig
+- Tool classification 
+- BM25 search
+- Bridge tool schemas
+- Assembly logic
+"""
+
+import pytest
+from praisonaiagents.tools.tool_search import (
+    ToolSearchConfig, classify_tools, search_catalog, bridge_tool_schemas,
+    assemble_tool_defs, dispatch_tool_search, dispatch_tool_describe,
+    resolve_underlying_call, PRAISONAI_CORE_TOOLS,
+    estimate_tool_schema_tokens, should_defer_tools
+)
+
+class TestToolSearchConfig:
+    """Test ToolSearchConfig creation and validation."""
+    
+    def test_default_config(self):
+        """Test default configuration."""
+        config = ToolSearchConfig()
+        assert config.enabled == "auto"
+        assert config.threshold_pct == 10.0
+        assert config.search_default_limit == 5
+        assert config.max_search_limit == 20
+        assert config.core_tools is None
+    
+    def test_from_raw_bool(self):
+        """Test creating config from boolean values."""
+        config_true = ToolSearchConfig.from_raw(True)
+        assert config_true.enabled == "on"
+        
+        config_false = ToolSearchConfig.from_raw(False)
+        assert config_false.enabled == "off"
+    
+    def test_from_raw_string(self):
+        """Test creating config from string values."""
+        config_auto = ToolSearchConfig.from_raw("auto")
+        assert config_auto.enabled == "auto"
+        
+        config_on = ToolSearchConfig.from_raw("on") 
+        assert config_on.enabled == "on"
+        
+        config_off = ToolSearchConfig.from_raw("off")
+        assert config_off.enabled == "off"
+    
+    def test_from_raw_dict(self):
+        """Test creating config from dictionary."""
+        config = ToolSearchConfig.from_raw({
+            "enabled": "on",
+            "threshold_pct": 15.0,
+            "search_default_limit": 10
+        })
+        assert config.enabled == "on"
+        assert config.threshold_pct == 15.0
+        assert config.search_default_limit == 10
+    
+    def test_from_raw_invalid(self):
+        """Test error handling for invalid inputs."""
+        with pytest.raises(ValueError):
+            ToolSearchConfig.from_raw("invalid")
+        
+        with pytest.raises(TypeError):
+            ToolSearchConfig.from_raw(123)
+
+class TestToolClassification:
+    """Test tool classification into core vs deferrable."""
+    
+    def test_core_tool_classification(self):
+        """Test that core tools are never deferred."""
+        tool_defs = [
+            {
+                "type": "function",
+                "function": {"name": "read_file", "description": "Read a file"}
+            },
+            {
+                "type": "function", 
+                "function": {"name": "search_web", "description": "Search the web"}
+            }
+        ]
+        
+        config = ToolSearchConfig()
+        core_tools, deferrable_tools = classify_tools(tool_defs, config)
+        
+        assert len(core_tools) == 2
+        assert len(deferrable_tools) == 0
+        assert core_tools == tool_defs
+    
+    def test_mcp_tool_classification(self):
+        """Test that MCP tools are marked as deferrable.""" 
+        tool_defs = [
+            {
+                "type": "function",
+                "function": {"name": "mcp_weather", "description": "Get weather"},
+                "__praisonai_deferrable__": True
+            },
+            {
+                "type": "function",
+                "function": {"name": "read_file", "description": "Read a file"}
+            }
+        ]
+        
+        config = ToolSearchConfig()
+        core_tools, deferrable_tools = classify_tools(tool_defs, config)
+        
+        assert len(core_tools) == 1
+        assert len(deferrable_tools) == 1
+        assert core_tools[0]["function"]["name"] == "read_file"
+        assert deferrable_tools[0]["function"]["name"] == "mcp_weather"
+    
+    def test_unknown_tool_stays_visible(self):
+        """Test that unknown tools stay visible (design invariant #2)."""
+        tool_defs = [
+            {
+                "type": "function",
+                "function": {"name": "unknown_tool", "description": "Unknown tool"}
+            }
+        ]
+        
+        config = ToolSearchConfig()
+        core_tools, deferrable_tools = classify_tools(tool_defs, config)
+        
+        # Unknown tools should stay in core (visible)
+        assert len(core_tools) == 1
+        assert len(deferrable_tools) == 0
+        assert core_tools[0]["function"]["name"] == "unknown_tool"
+
+class TestBM25Search:
+    """Test BM25 search functionality."""
+    
+    def test_empty_search(self):
+        """Test search with empty inputs."""
+        result = search_catalog([], "test query")
+        assert result == []
+        
+        result = search_catalog([{"type": "function", "function": {"name": "test"}}], "")
+        assert result == []
+    
+    def test_simple_search(self):
+        """Test basic search functionality."""
+        deferrable_tools = [
+            {
+                "type": "function",
+                "function": {
+                    "name": "weather_tool",
+                    "description": "Get current weather information"
+                }
+            },
+            {
+                "type": "function", 
+                "function": {
+                    "name": "stock_tool",
+                    "description": "Get stock market data"
+                }
+            }
+        ]
+        
+        results = search_catalog(deferrable_tools, "weather", limit=10)
+        assert len(results) >= 1
+        assert any("weather_tool" in r["name"] for r in results)
+        
+        results = search_catalog(deferrable_tools, "stock", limit=10)
+        assert len(results) >= 1
+        assert any("stock_tool" in r["name"] for r in results)
+
+class TestBridgeTools:
+    """Test bridge tool schemas and assembly."""
+    
+    def test_bridge_schemas(self):
+        """Test bridge tool schema generation."""
+        schemas = bridge_tool_schemas()
+        assert len(schemas) == 3
+        
+        tool_names = {schema["function"]["name"] for schema in schemas}
+        assert tool_names == {"tool_search", "tool_describe", "tool_call"}
+        
+        # Validate schema structure
+        for schema in schemas:
+            assert schema["type"] == "function"
+            assert "function" in schema
+            assert "name" in schema["function"]
+            assert "description" in schema["function"] 
+            assert "parameters" in schema["function"]
+    
+    def test_assemble_disabled(self):
+        """Test assembly when tool search is disabled."""
+        tool_defs = [
+            {"type": "function", "function": {"name": "test_tool", "description": "Test"}}
+        ]
+        
+        config = ToolSearchConfig(enabled="off")
+        assembled, metadata = assemble_tool_defs(tool_defs, config)
+        
+        assert assembled == tool_defs
+        assert metadata["bridge_mode"] is False
+        assert metadata["deferred_count"] == 0
+    
+    def test_assemble_no_deferrable_tools(self):
+        """Test assembly when no tools are deferrable."""
+        tool_defs = [
+            {"type": "function", "function": {"name": "read_file", "description": "Read file"}}
+        ]
+        
+        config = ToolSearchConfig(enabled="on")
+        assembled, metadata = assemble_tool_defs(tool_defs, config)
+        
+        # Should return original tools since nothing to defer
+        assert assembled == tool_defs
+        assert metadata["bridge_mode"] is False
+    
+    def test_assemble_with_deferrable_tools(self):
+        """Test assembly with deferrable tools."""
+        tool_defs = [
+            {"type": "function", "function": {"name": "read_file", "description": "Read file"}},
+            {
+                "type": "function", 
+                "function": {"name": "mcp_weather", "description": "Get weather"},
+                "__praisonai_deferrable__": True
+            }
+        ]
+        
+        config = ToolSearchConfig(enabled="on")
+        assembled, metadata = assemble_tool_defs(tool_defs, config)
+        
+        # Should have core tool + 3 bridge tools
+        assert len(assembled) == 4
+        assert metadata["bridge_mode"] is True
+        assert metadata["deferred_count"] == 1
+        assert len(metadata["catalog"]) == 1
+        
+        # Verify core tool remains
+        core_tool_names = {t["function"]["name"] for t in assembled if t["function"]["name"] != "tool_search" and t["function"]["name"] != "tool_describe" and t["function"]["name"] != "tool_call"}
+        assert "read_file" in core_tool_names
+
+class TestBridgeDispatchers:
+    """Test bridge tool dispatch handlers."""
+    
+    def test_tool_search_dispatch(self):
+        """Test tool_search bridge dispatch."""
+        deferrable_tools = [
+            {
+                "type": "function",
+                "function": {
+                    "name": "weather_tool", 
+                    "description": "Get weather information"
+                }
+            }
+        ]
+        
+        config = ToolSearchConfig()
+        result = dispatch_tool_search("weather", 5, deferrable_tools, config)
+        
+        assert "query" in result
+        assert "results" in result
+        assert "total_available" in result
+        assert result["query"] == "weather"
+        assert result["total_available"] == 1
+    
+    def test_tool_describe_dispatch(self):
+        """Test tool_describe bridge dispatch."""
+        deferrable_tools = [
+            {
+                "type": "function",
+                "function": {
+                    "name": "weather_tool",
+                    "description": "Get weather information"
+                }
+            }
+        ]
+        
+        result = dispatch_tool_describe("weather_tool", deferrable_tools)
+        
+        assert "tool_name" in result
+        assert "found" in result
+        assert result["found"] is True
+        assert result["tool_name"] == "weather_tool"
+        assert "schema" in result
+    
+    def test_tool_describe_not_found(self):
+        """Test tool_describe for non-existent tool."""
+        result = dispatch_tool_describe("nonexistent_tool", [])
+        
+        assert "tool_name" in result
+        assert "found" in result
+        assert result["found"] is False
+        assert "error" in result
+
+class TestToolCallUnwrapping:
+    """Test tool_call bridge unwrapping."""
+    
+    def test_resolve_underlying_call(self):
+        """Test unwrapping tool_call bridge."""
+        # Normal tool call - should pass through
+        name, args = resolve_underlying_call("normal_tool", {"arg1": "value1"})
+        assert name == "normal_tool"
+        assert args == {"arg1": "value1"}
+        
+        # Bridge tool call - should unwrap
+        bridge_args = {
+            "tool_name": "real_tool",
+            "tool_args": {"param1": "test"}
+        }
+        name, args = resolve_underlying_call("tool_call", bridge_args)
+        assert name == "real_tool"
+        assert args == {"param1": "test"}
+    
+    def test_resolve_missing_tool_name(self):
+        """Test error handling for missing tool_name."""
+        with pytest.raises(ValueError, match="tool_call requires 'tool_name' parameter"):
+            resolve_underlying_call("tool_call", {"tool_args": {}})
+
+class TestTokenEstimation:
+    """Test token estimation and threshold logic."""
+    
+    def test_estimate_tool_schema_tokens(self):
+        """Test token estimation for tool schemas.""" 
+        tool_defs = [
+            {
+                "type": "function",
+                "function": {
+                    "name": "test_tool",
+                    "description": "A test tool with parameters",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {"param1": {"type": "string"}}
+                    }
+                }
+            }
+        ]
+        
+        tokens = estimate_tool_schema_tokens(tool_defs)
+        assert tokens > 0
+        assert isinstance(tokens, int)
+    
+    def test_should_defer_tools(self):
+        """Test deferral threshold logic."""
+        # Create some dummy deferrable tools
+        deferrable_tools = [
+            {
+                "type": "function",
+                "function": {"name": f"tool_{i}", "description": "Test tool"}
+            } 
+            for i in range(10)
+        ]
+        
+        # Test with disabled config
+        config_off = ToolSearchConfig(enabled="off")
+        assert should_defer_tools(deferrable_tools, config_off, 20000) is False
+        
+        # Test with on config 
+        config_on = ToolSearchConfig(enabled="on")
+        assert should_defer_tools(deferrable_tools, config_on, 20000) is True
+        
+        # Test auto mode with high threshold (shouldn't defer)
+        config_auto_high = ToolSearchConfig(enabled="auto", threshold_pct=90)
+        assert should_defer_tools(deferrable_tools, config_auto_high, 20000) is False
+        
+        # Test auto mode with low threshold (should defer)
+        config_auto_low = ToolSearchConfig(enabled="auto", threshold_pct=1)  
+        assert should_defer_tools(deferrable_tools, config_auto_low, 20000) is True
+
+class TestCoreToolsInvariant:
+    """Test that core tools never defer."""
+    
+    def test_core_tools_constant(self):
+        """Test that PRAISONAI_CORE_TOOLS is properly defined."""
+        assert isinstance(PRAISONAI_CORE_TOOLS, frozenset)
+        assert len(PRAISONAI_CORE_TOOLS) > 0
+        
+        # Check some expected core tools
+        expected_core = {"read_file", "execute_command", "search_web", "clarify"}
+        assert expected_core.issubset(PRAISONAI_CORE_TOOLS)
+    
+    def test_core_tools_never_deferred(self):
+        """Test that core tools are never put in deferrable category."""
+        # Create tools that are marked as deferrable but are actually core tools
+        tool_defs = []
+        for tool_name in list(PRAISONAI_CORE_TOOLS)[:5]:  # Test first 5 core tools
+            tool_defs.append({
+                "type": "function",
+                "function": {"name": tool_name, "description": f"Test {tool_name}"},
+                "__praisonai_deferrable__": True  # Try to mark as deferrable
+            })
+        
+        config = ToolSearchConfig()
+        core_tools, deferrable_tools = classify_tools(tool_defs, config)
+        
+        # All should end up in core tools despite the deferrable marker
+        assert len(core_tools) == 5
+        assert len(deferrable_tools) == 0

src/praisonai-agents/tests/unit/tools/test_tool_search_simple.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""
+Simple test script to verify tool search implementation.
+"""
+
+import sys
+
+
+def test_tool_search_basic():
+    """Test basic tool search functionality."""
+    print("Testing Tool Search implementation...")
+
+    try:
+        # Test config creation
+        from praisonaiagents.tools.tool_search import ToolSearchConfig
+        config = ToolSearchConfig()
+        print(f"✓ ToolSearchConfig created: enabled={config.enabled}, threshold={config.threshold_pct}%")
+
+        # Test config from raw values
+        config_true = ToolSearchConfig.from_raw(True)
+        config_false = ToolSearchConfig.from_raw(False)
+        print(f"✓ Config from bool: True→{config_true.enabled}, False→{config_false.enabled}")
+
+        # Test tool classification
+        from praisonaiagents.tools.tool_search import PRAISONAI_CORE_TOOLS
+        print(f"✓ Core tools defined: {len(PRAISONAI_CORE_TOOLS)} tools")
+        print(f"  Sample core tools: {list(PRAISONAI_CORE_TOOLS)[:5]}")
+
+        # Test bridge tools
+        from praisonaiagents.tools.tool_search import bridge_tool_schemas
+        bridge_schemas = bridge_tool_schemas()
+        bridge_names = [schema["function"]["name"] for schema in bridge_schemas]
+        print(f"✓ Bridge tools created: {bridge_names}")
+
+        # Test assembly logic
+        from praisonaiagents.tools.tool_search import assemble_tool_defs
+
+        # Test with disabled config
+        tool_defs = [
+            {"type": "function", "function": {"name": "test_tool", "description": "Test"}}
+        ]
+        assembled, metadata = assemble_tool_defs(tool_defs, ToolSearchConfig(enabled="off"))
+        print(f"✓ Disabled mode: bridge={metadata['bridge_mode']}, deferred={metadata['deferred_count']}")
+
+        # Test with enabled but no deferrable tools
+        assembled, metadata = assemble_tool_defs(tool_defs, ToolSearchConfig(enabled="on"))
+        print(f"✓ No deferrable tools: bridge={metadata['bridge_mode']}, deferred={metadata['deferred_count']}")
+
+        # Test with deferrable tools
+        deferrable_tool_defs = [
+            {"type": "function", "function": {"name": "read_file", "description": "Read file"}},
+            {
+                "type": "function",
+                "function": {"name": "mcp_weather", "description": "Get weather"},
+                "__praisonai_deferrable__": True
+            }
+        ]
+        assembled, metadata = assemble_tool_defs(deferrable_tool_defs, ToolSearchConfig(enabled="on"))
+        print(f"✓ With deferrable tools: bridge={metadata['bridge_mode']}, deferred={metadata['deferred_count']}")
+        print(f"  Assembled tools: {len(assembled)}, catalog: {len(metadata['catalog'])}")
+
+        # Test search functionality
+        from praisonaiagents.tools.tool_search import search_catalog
+        search_results = search_catalog(metadata["deferrable_tools"], "weather", limit=5)
+        print(f"✓ Search 'weather': {len(search_results)} results")
+        if search_results:
+            print(f"  First result: {search_results[0]}")
+
+        print("\n🎉 All basic tests passed! Tool Search implementation is working.")
+        return True
+
+    except Exception as e:
+        print(f"❌ Test failed: {e}")
+        import traceback
+        traceback.print_exc()
+        return False
+
+
+def test_agent_integration():
+    """Test Agent class integration."""
+    print("\nTesting Agent integration...")
+
+    try:
+        # Test config classes are exported
+        from praisonaiagents.config import ToolSearchConfig
+        print("✓ ToolSearchConfig available in config module")
+
+        # Test Agent parameter (basic import check)
+        from praisonaiagents.agent.agent import Agent
+        print("✓ Agent class imports successfully")
+
+        print("✓ Agent integration looks good")
+        return True
+
+    except Exception as e:
+        print(f"❌ Agent integration test failed: {e}")
+        import traceback
+        traceback.print_exc()
+        return False
+
+
+if __name__ == "__main__":
+    print("Tool Search Implementation Test")
+    print("=" * 50)
+
+    success1 = test_tool_search_basic()
+    success2 = test_agent_integration()
+
+    if success1 and success2:
+        print("\n✅ All tests passed!")
+        sys.exit(0)
+    else:
+        print("\n❌ Some tests failed!")
+        sys.exit(1)

src/praisonai-agents/uv.lock

@@ -3125,7 +3125,7 @@ wheels = [
 
 [[package]]
 name = "praisonaiagents"
-version = "1.6.48"
+version = "1.6.52"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },

src/praisonai-platform/praisonai_platform/services/auth_service.py

@@ -23,15 +23,37 @@ from ..db.models import Member, User
 _pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
 _DEFAULT_SECRET = "dev-secret-change-me"
-JWT_SECRET = os.environ.get("PLATFORM_JWT_SECRET", _DEFAULT_SECRET)
+JWT_SECRET = os.environ.get("PLATFORM_JWT_SECRET")
 JWT_ALGORITHM = "HS256"
 JWT_TTL_SECONDS = int(os.environ.get("PLATFORM_JWT_TTL", str(30 * 24 * 3600)))
 
-if JWT_SECRET == _DEFAULT_SECRET and os.environ.get("PLATFORM_ENV", "dev") != "dev":
-    raise RuntimeError(
-        "PLATFORM_JWT_SECRET must be set to a strong random value in production. "
-        "Set PLATFORM_ENV=dev to suppress this check during development."
-    )
+if JWT_SECRET is None or JWT_SECRET == _DEFAULT_SECRET:
+    # Determine if we are in an explicit dev environment.
+    # Only trust PLATFORM_ENV when it is explicitly set — an unset value
+    # must NOT silently fall back to "dev" because production deployments
+    # that forget the variable would run with a well-known signing key.
+    _explicit_env = os.environ.get("PLATFORM_ENV")
+    if _explicit_env == "dev":
+        # Developer explicitly opted in to insecure defaults — allow it.
+        JWT_SECRET = _DEFAULT_SECRET
+    else:
+        # Either production or environment not specified.
+        # Auto-generate an ephemeral secret so the service can start, but
+        # warn loudly because tokens will not survive a restart.
+        import secrets as _secrets
+        import warnings as _warnings
+
+        JWT_SECRET = _secrets.token_urlsafe(32)
+        _warnings.warn(
+            "PLATFORM_JWT_SECRET is not set. Auto-generated an ephemeral "
+            "signing key — JWT tokens will be invalidated on restart. "
+            "Set PLATFORM_JWT_SECRET to a stable random value for production "
+            "use, or set PLATFORM_ENV=dev to use the development default.",
+            stacklevel=2,
+        )
+
+# Type-narrow: at this point JWT_SECRET is always a non-None str.
+assert JWT_SECRET is not None  # noqa: S105
 
 
 class AuthService:
@@ -113,10 +135,6 @@ class AuthService:
 
     def _issue_token(self, user: User) -> str:
         """Issue a JWT for a user."""
-        if JWT_SECRET == _DEFAULT_SECRET and os.environ.get("PLATFORM_ENV", "dev") != "dev":
-            raise RuntimeError(
-                "Refusing to issue JWT with default PLATFORM_JWT_SECRET outside dev"
-            )
         now = datetime.now(timezone.utc)
         payload = {
             "sub": user.id,

src/praisonai-rust/FEATURE_PARITY_TRACKER.json

@@ -1,14 +1,14 @@
 {
   "version": "1.5.87",
-  "lastUpdated": "2026-05-30",
+  "lastUpdated": "2026-06-03",
   "generatedBy": "praisonai._dev.parity.generator",
   "sourceOfTruth": "Python SDK (praisonaiagents)",
   "status": "PARITY_ACHIEVED",
   "summary": {
-    "pythonCoreFeatures": 335,
+    "pythonCoreFeatures": 336,
     "rustFeatures": 667,
-    "gapCount": 56,
-    "parityPercentage": 199.1
+    "gapCount": 57,
+    "parityPercentage": 198.5
   },
   "pythonCoreSDK": {
     "exports": [
@@ -245,6 +245,7 @@
       "ToolPluginProtocol",
       "ToolRegistry",
       "ToolResult",
+      "ToolSearchConfig",
       "ToolUseEvent",
       "ToolValidationError",
       "Tools",
@@ -4161,6 +4162,15 @@
         "status": "DONE",
         "category": "plugins"
       },
+      {
+        "feature": "ToolSearchConfig",
+        "python": true,
+        "rust": false,
+        "priority": "P3",
+        "effort": "low",
+        "status": "TODO",
+        "category": "config"
+      },
       {
         "feature": "ToolUseEvent",
         "python": true,

src/praisonai-rust/PARITY.md

@@ -1,16 +1,16 @@
 # Rust Feature Parity Tracker
 
-> **Python Features:** 335 | **Rust Features:** 667 | **Parity:** 85.1%
+> **Python Features:** 336 | **Rust Features:** 667 | **Parity:** 84.8%
 
 ## Summary
 
 | Metric | Count |
 |--------|-------|
-| Python Core Features | 335 |
+| Python Core Features | 336 |
 | Rust Features | 667 |
-| **Actual Gap Count** | **50** |
+| **Actual Gap Count** | **51** |
 | Language Limitations (N/A) | 5 |
-| **Parity** | **85.1%** |
+| **Parity** | **84.8%** |
 
 ## Implemented Features
 
@@ -735,6 +735,7 @@ These Python features cannot be directly implemented in Rust due to reserved key
 - ❌ `StopReason`
 - ❌ `StructuredFormatter`
 - ❌ `ToolExecutionError`
+- ❌ `ToolSearchConfig`
 - ❌ `ToolUseEvent`
 - ❌ `ToolValidationError`
 - ❌ `configure_structured_logging`

src/praisonai-ts/FEATURE_PARITY_TRACKER.json

@@ -1,17 +1,17 @@
 {
   "version": "1.5.87",
-  "lastUpdated": "2026-05-30",
+  "lastUpdated": "2026-06-03",
   "generatedBy": "praisonai._dev.parity.generator",
   "sourceOfTruth": "Python SDK (praisonaiagents)",
   "summary": {
-    "pythonCoreFeatures": 335,
+    "pythonCoreFeatures": 336,
     "pythonWrapperFeatures": 99,
     "typescriptFeatures": 1192,
-    "gapCount": 43,
+    "gapCount": 44,
     "priorityP0": 1,
     "priorityP1": 0,
     "priorityP2": 0,
-    "priorityP3": 42
+    "priorityP3": 43
   },
   "pythonCoreSDK": {
     "path": "/home/runner/work/PraisonAI/PraisonAI/src/praisonai-agents/praisonaiagents",
@@ -249,6 +249,7 @@
       "ToolPluginProtocol",
       "ToolRegistry",
       "ToolResult",
+      "ToolSearchConfig",
       "ToolUseEvent",
       "ToolValidationError",
       "Tools",
@@ -475,6 +476,7 @@
         "ReflectionConfig",
         "SkillsConfig",
         "TemplateConfig",
+        "ToolSearchConfig",
         "WEB_PRESETS",
         "WebConfig",
         "WebSearchProvider",
@@ -5254,6 +5256,15 @@
         "status": "DONE",
         "category": "plugins"
       },
+      {
+        "feature": "ToolSearchConfig",
+        "python": true,
+        "typescript": false,
+        "priority": "P3",
+        "effort": "low",
+        "status": "TODO",
+        "category": "config"
+      },
       {
         "feature": "ToolUseEvent",
         "python": true,

src/praisonai-ts/PARITY.md

@@ -1,20 +1,20 @@
 # Feature Parity Tracker
 
-> **Version:** 1.5.87 | **Last Updated:** 2026-05-30
+> **Version:** 1.5.87 | **Last Updated:** 2026-06-03
 > **Source of Truth:** Python SDK (praisonaiagents)
 
 ## Summary
 
 | Metric | Count |
 |--------|-------|
-| Python Core Features | 335 |
+| Python Core Features | 336 |
 | Python Wrapper Features | 99 |
 | TypeScript Features | 1192 |
-| **Gap Count** | **43** |
+| **Gap Count** | **44** |
 | P0 (Critical) | 1 |
 | P1 (High) | 0 |
 | P2 (Medium) | 0 |
-| P3 (Low) | 42 |
+| P3 (Low) | 43 |
 
 ## Gap Matrix
 
@@ -147,7 +147,7 @@
 | `enable\_telemetry` | ✅ | ✅ | low | ✅ DONE |
 | `get\_telemetry` | ✅ | ✅ | low | ✅ DONE |
 
-### P3_Advanced (179 done, 42 todo)
+### P3_Advanced (179 done, 43 todo)
 
 | Feature | Python | TypeScript | Effort | Status |
 |---------|--------|------------|--------|--------|
@@ -189,6 +189,7 @@
 | `StopReason` | ✅ | ❌ | high | ⏳ TODO |
 | `StructuredFormatter` | ✅ | ❌ | high | ⏳ TODO |
 | `ToolExecutionError` | ✅ | ❌ | low | ⏳ TODO |
+| `ToolSearchConfig` | ✅ | ❌ | low | ⏳ TODO |
 | `ToolUseEvent` | ✅ | ❌ | high | ⏳ TODO |
 | `ValidationError` | ✅ | ❌ | low | ⏳ TODO |
 | `configure\_structured\_logging` | ✅ | ❌ | low | ⏳ TODO |
@@ -423,7 +424,7 @@ from praisonaiagents import ConditionProtocol, DictCondition, ExpressionConditio
 </details>
 
 <details>
-<summary><strong>config</strong> (75 exports)</summary>
+<summary><strong>config</strong> (76 exports)</summary>
 
 ```python
 from praisonaiagents import AUTONOMY_PRESETS, ArrayMode, AutonomyLevel, CACHING_PRESETS, CONTEXT_PRESETS, CachingConfig, ChunkingStrategy, ConfigValidationError, DefaultsConfig, EXECUTION_PRESETS...

src/praisonai/README.md

@@ -382,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)
 

src/praisonai/examples/recipe_runtime_example.yaml

@@ -85,7 +85,7 @@ examples:
   - name: Basic usage
     command: praisonai recipe run news-monitor
   - name: With custom topic
-    command: praisonai recipe run news-monitor -i '{"topic": "LLMs"}'
+    command: 'praisonai recipe run news-monitor -i ''{"topic": "LLMs"}'''
   - name: Background execution
     command: praisonai recipe run news-monitor --background
   - name: Submit as job