fix(examples): phase-3 API alignment for consolidated params on v4.6.52 (#1864)

Co-authored-by: Dhivya-Bharathy <[email protected]>

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

@@ -2,7 +2,7 @@
 from praisonaiagents import Agent
 from praisonaiagents import AgentFlow, Task
 from praisonaiagents.workflows.workflow_configs import (
-    WorkflowOutputConfig, WorkflowPlanningConfig, WorkflowMemoryConfig,
+    WorkflowPlanningConfig, WorkflowMemoryConfig,
 )
 
 # Full-featured workflow with all consolidated params
@@ -32,7 +32,7 @@ workflow = AgentFlow(
         ),
     ],
     # Workflow-level consolidated params
-    output=WorkflowOutputConfig(output="verbose", stream=True),
+    output="verbose",
     planning=WorkflowPlanningConfig(enabled=True, reasoning=True),
     memory=WorkflowMemoryConfig(backend="file"),
     # Agent-like params

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()
     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

@@ -48,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
@@ -70,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/managed-agents/persistence/cockroachdb_managed.py

@@ -17,8 +17,10 @@ from praisonaiagents import Agent
 agent = Agent(
     name="CockroachDB Agent",
     instructions="You are a helpful assistant.",
-    db={"database_url": os.getenv("COCKROACHDB_URL")},
-    session_id="crdb-demo-session",
+    memory={
+        "session_id": "crdb-demo-session",
+        "db": {"database_url": os.getenv("COCKROACHDB_URL")},
+    },
 )
 
 result = agent.start("What is distributed SQL? Explain in 2 sentences.")

examples/managed-agents/persistence/neon_managed.py

@@ -18,8 +18,10 @@ from praisonaiagents import Agent
 agent = Agent(
     name="Neon Agent",
     instructions="You are a helpful assistant.",
-    db={"database_url": os.getenv("NEON_DATABASE_URL")},
-    session_id="neon-demo-session",
+    memory={
+        "session_id": "neon-demo-session",
+        "db": {"database_url": os.getenv("NEON_DATABASE_URL")},
+    },
 )
 
 # Option 2: Direct URL

examples/managed-agents/persistence/supabase_direct_managed.py

@@ -20,8 +20,10 @@ from praisonaiagents import Agent
 agent = Agent(
     name="Supabase Agent",
     instructions="You are a helpful assistant.",
-    db={"database_url": os.getenv("SUPABASE_DATABASE_URL")},
-    session_id="supabase-demo-session",
+    memory={
+        "session_id": "supabase-demo-session",
+        "db": {"database_url": os.getenv("SUPABASE_DATABASE_URL")},
+    },
 )
 
 result = agent.start("What is Supabase? Explain in 2 sentences.")

examples/managed-agents/persistence/turso_managed.py

@@ -18,11 +18,13 @@ from praisonaiagents import Agent
 agent = Agent(
     name="Turso Agent",
     instructions="You are a helpful assistant.",
-    db={
-        "database_url": os.getenv("TURSO_DATABASE_URL"),
-        "auth_token": os.getenv("TURSO_AUTH_TOKEN"),
+    memory={
+        "session_id": "turso-demo-session",
+        "db": {
+            "database_url": os.getenv("TURSO_DATABASE_URL"),
+            "auth_token": os.getenv("TURSO_AUTH_TOKEN"),
+        },
     },
-    session_id="turso-demo-session",
 )
 
 result = agent.start("What is edge computing? Explain in 2 sentences.")

examples/middleware/configurable_model.py

@@ -10,8 +10,7 @@ from praisonaiagents import Agent
 agent = Agent(
     name="FlexBot",
     instructions="You are a helpful assistant.",
-    llm="gpt-4o-mini",
-    llm_config={"configurable": True}
+    llm={"model": "gpt-4o-mini", "configurable": True},
 )
 
 if __name__ == "__main__":

examples/middleware/injected_state.py

@@ -21,7 +21,7 @@ agent = Agent(
     name="ContextBot",
     instructions="You help show context information.",
     tools=[show_context],
-    session_id="my-session-123"
+    memory={"session_id": "my-session-123"},
 )
 
 if __name__ == "__main__":

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/policy/basic_policy.py

@@ -57,8 +57,8 @@ def main():
     agent = Agent(
         name="SecureAgent",
         instructions="You are a file management assistant.",
-        policy=engine
     )
+    agent.policy = engine
     
     print("\n--- Agent with Policy Enforcement Created ---")
     print(f"Agent: {agent.name}")

examples/python/camera/camera-continuous.py

@@ -75,7 +75,7 @@ class ContinuousCameraMonitor:
             agents=[self.vision_agent],
             tasks=[task],
             process="sequential",
-            verbose=0  # Reduced verbosity for continuous monitoring
+            output="silent",  # Reduced verbosity for continuous monitoring
         )
         
         return agents.start()

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,8 @@
-from praisonaiagents import Agent, Task, AgentTeam
-import logging
+import tempfile
 import os
+import sys
+import logging
+from praisonaiagents import Agent, Task, AgentTeam
 
 # Configure logging
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
@@ -17,14 +19,20 @@ config = {
     }
 }
 
+# Create sample file in system temp directory
+_sample_path = os.path.join(tempfile.gettempdir(), "sample_knowledge.txt")
+if not os.path.exists(_sample_path):
+    with open(_sample_path, "w", encoding="utf-8") as f:
+        f.write("Mervin Praison is the creator of PraisonAI.\n")
+
 # 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": [_sample_path]},
+)
 
 # Define a task for the agent
 knowledge_task = Task(
@@ -39,8 +47,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/concepts/repetitive-agents.py

@@ -16,7 +16,7 @@ agents = AgentTeam(
     agents=[agent],
     tasks=[task],
     process="workflow",
-    max_iter=30
+    execution={"max_iter": 30},
 )
 
 agents.start()

examples/python/custom_tools/example_latency_tracking.py

@@ -61,7 +61,6 @@ def example_with_wrappers():
         role="Data Analyst",
         goal="Analyze data and provide insights",
         llm="gpt-4o-mini",
-        request_id="request_2"  # Unique request ID
     )
     
     # Operations are automatically tracked

examples/python/data/cot-agents.py

@@ -132,7 +132,8 @@ agents = AgentTeam(
     agents=[qa_generator, total_questions_evaluator, cot_generator, upload_to_huggingface],
     tasks=[generate_task, evaluate_total_questions_task, generate_cot_task, upload_to_huggingface_task],
     process="workflow",
-    max_iter=30, output="minimal"
+    execution={"max_iter": 30},
+    output="minimal",
 )
 
 agents.start()

examples/python/failover_example.py

@@ -18,7 +18,7 @@ openai_profile = AuthProfile(
     provider="openai",
     api_key=os.environ.get("OPENAI_API_KEY", "sk-..."),
     priority=1,  # Highest priority
-    rate_limit=100,
+    rate_limit_rpm=100,
 )
 
 anthropic_profile = AuthProfile(
@@ -26,7 +26,7 @@ anthropic_profile = AuthProfile(
     provider="anthropic",
     api_key=os.environ.get("ANTHROPIC_API_KEY", "sk-ant-..."),
     priority=2,  # Second priority
-    rate_limit=50,
+    rate_limit_rpm=50,
 )
 
 groq_profile = AuthProfile(
@@ -34,7 +34,7 @@ groq_profile = AuthProfile(
     provider="groq",
     api_key=os.environ.get("GROQ_API_KEY", "gsk-..."),
     priority=3,  # Third priority (fast but limited)
-    rate_limit=30,
+    rate_limit_rpm=30,
 )
 
 # Configure failover behavior

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/guardrails/production-guardrails-patterns.py

@@ -6,6 +6,7 @@ built-in validation and safety mechanisms for secure agent operations.
 """
 
 from praisonaiagents import Agent, Task, AgentTeam
+from praisonaiagents.config.feature_configs import GuardrailConfig
 from praisonaiagents.tools import internet_search
 
 print("=== Production Guardrails Patterns Example ===\n")
@@ -38,8 +39,8 @@ production_agent = Agent(
     goal="Generate safe, compliant content for production use",
     backstory="Production-ready agent with built-in safety and compliance checks",
     tools=[internet_search],
-    guardrails=content_guardrail,
-    max_retries=3)
+    guardrails=GuardrailConfig(validator=content_guardrail, max_retries=3),
+)
 
 # Create secure task with validation
 secure_task = Task(

examples/python/linear_agent_example.py

@@ -40,7 +40,7 @@ agent = Agent(
         "execute_command"
     ],
     memory=True,
-    web_search=True,
+    web=True,
     auto_approve_tools=True,
 )
 

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/models/googlegemini/any-llm-agents-advanced.py

@@ -31,10 +31,9 @@ llm_config = {
 agent = Agent(
     instructions="You are a helpful Assistant specialized in scientific explanations. "
                 "Provide clear, accurate, and engaging responses.",
-    llm=llm_config,                  # Pass the detailed configuration
-    # Enable detailed output
-    markdown=True,                   # Format responses in markdown
-    reflection=True               # Enable self-reflection
+    llm=llm_config,
+    output={"markdown": True},
+    reflection=True,
 )
 
 # Test the agent

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

@@ -12,6 +12,7 @@ Environment:
 """
 
 import os
+import tempfile
 from praisonaiagents import Agent, AutoRagAgent
 
 # Ensure API key is set
@@ -20,8 +21,8 @@ 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:
+sample_doc = os.path.join(tempfile.gettempdir(), "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/session/session_persistence_example.py

@@ -28,7 +28,7 @@ def example_basic_persistence():
     agent = Agent(
         name="Assistant",
         instructions="You are a helpful assistant. Be very brief.",
-        session_id=session_id,
+        memory={"session_id": session_id},
         output="minimal",
     )
     
@@ -59,7 +59,7 @@ def example_session_restoration():
     agent1 = Agent(
         name="MemoryBot",
         instructions="You are a helpful assistant. Be very brief.",
-        session_id=session_id,
+        memory={"session_id": session_id},
         output="minimal",
     )
     
@@ -72,7 +72,7 @@ def example_session_restoration():
     agent2 = Agent(
         name="MemoryBot",
         instructions="You are a helpful assistant. Be very brief.",
-        session_id=session_id,
+        memory={"session_id": session_id},
         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/tools/exa-tool/rag_examples/agentic_rag/rag_agent.py

@@ -22,7 +22,8 @@ rag_agent = Agent(
     
     You can use the internet_search function to search the web when needed.""",
     llm="gpt-4o",
-    markdown=True)
+    output={"markdown": True},
+)
 
 if __name__ == "__main__":
     print("🤖 Thai Recipe RAG Agent is ready!")

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_mixed_steps.py

@@ -53,7 +53,7 @@ workflow = AgentFlow(
         # Step 4: Function - formats output
         format_output
     ],
-    default_llm="gpt-4o-mini"  # Used for action-based steps
+    llm="gpt-4o-mini",  # Used for action-based steps
 )
 
 if __name__ == "__main__":

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/retrieval/basic_retrieval.py

@@ -51,13 +51,15 @@ print("=" * 60)
 agent = Agent(
     name="Knowledge Agent",
     instructions="You are a helpful assistant that answers questions based on the provided knowledge. Be concise.",
-    knowledge=[".praison/test_docs/overview.txt"],
-    retrieval_config={
-        "policy": "always",      # Always retrieve for this demo
-        "top_k": 3,              # Get top 3 chunks
-        "citations": True,       # Include citations
-        "max_context_tokens": 2000,
-    }
+    knowledge={
+        "sources": [".praison/test_docs/overview.txt"],
+        "config": {
+            "policy": "always",
+            "top_k": 3,
+            "citations": True,
+            "max_context_tokens": 2000,
+        },
+    },
 )
 
 # Test 1: Basic chat with retrieval

examples/skills/basic_skill_usage.py

@@ -60,7 +60,7 @@ print("=" * 50)
 agent_with_discovery = Agent(
     name="Multi-Skill Agent",
     instructions="You are a versatile assistant with multiple skills.",
-    skills_dirs=["./"],  # Scan current directory for skill subdirectories
+    skills=["./"],  # Scan current directory for skill subdirectories
 )
 
 if agent_with_discovery.skill_manager:

examples/yaml/workflows/example_usage.py

@@ -93,7 +93,7 @@ def example_3_execute_yaml_directly():
     result = manager.execute_yaml(
         yaml_file,
         input_data="The future of AI",
-        output="verbose"
+        verbose=True,
     )
     
     print(f"Status: {result['status']}")

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}"
             

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-rust/FEATURE_PARITY_TRACKER.json

@@ -1,6 +1,6 @@
 {
   "version": "1.5.87",
-  "lastUpdated": "2026-06-02",
+  "lastUpdated": "2026-06-03",
   "generatedBy": "praisonai._dev.parity.generator",
   "sourceOfTruth": "Python SDK (praisonaiagents)",
   "status": "PARITY_ACHIEVED",

src/praisonai-ts/FEATURE_PARITY_TRACKER.json

@@ -1,6 +1,6 @@
 {
   "version": "1.5.87",
-  "lastUpdated": "2026-06-02",
+  "lastUpdated": "2026-06-03",
   "generatedBy": "praisonai._dev.parity.generator",
   "sourceOfTruth": "Python SDK (praisonaiagents)",
   "summary": {

src/praisonai-ts/PARITY.md

@@ -1,6 +1,6 @@
 # Feature Parity Tracker
 
-> **Version:** 1.5.87 | **Last Updated:** 2026-06-02
+> **Version:** 1.5.87 | **Last Updated:** 2026-06-03
 > **Source of Truth:** Python SDK (praisonaiagents)
 
 ## Summary

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

src/praisonai/praisonai/_entrypoint.py

@@ -65,9 +65,32 @@ async def arun(agent_file: str,
                tools: list | None = None,
                agent_yaml: str | None = None,
                cli_config: dict | None = None) -> str:
-    """Async equivalent of `run()` for FastAPI / Jupyter async callers."""
-    import asyncio
-    return await asyncio.to_thread(
-        run, agent_file, framework,
-        tools=tools, agent_yaml=agent_yaml, cli_config=cli_config,
-    )
+    """Async equivalent of `run()` using native async framework adapters."""
+    from .agents_generator import AgentsGenerator
+    from ._framework_availability import is_available
+
+    if framework is None:
+        # Mirror the CLI default-resolution order
+        for candidate in ("crewai", "praisonaiagents", "autogen", "ag2"):
+            if is_available(candidate):
+                framework = "praisonai" if candidate == "praisonaiagents" else candidate
+                break
+        else:
+            raise RuntimeError(
+                "No supported framework installed. "
+                "Install one of: crewai, praisonaiagents, autogen, ag2."
+            )
+    else:
+        # Validate explicit framework like CLI does
+        from .framework_adapters.validators import assert_framework_available
+        assert_framework_available(framework)
+
+    gen = AgentsGenerator(
+        agent_file=agent_file,
+        framework=framework,
+        config_list=_build_config_list(),
+        tools=tools,
+        agent_yaml=agent_yaml,
+        cli_config=cli_config,
+    )
+    return await gen.agenerate_crew_and_kickoff()

src/praisonai/praisonai/agent.py

@@ -24,39 +24,23 @@ class Agent(CoreAgent):
             *args: Positional arguments for core Agent
             cli_backend: CLI backend for delegating full turns. Accepts:
                 - str: Backend ID ("claude-code", "codex-cli") - resolved in wrapper
+                - dict: Backend config {"id": "claude-code", "overrides": {...}}
                 - CliBackendProtocol: Pre-resolved backend instance
                 - callable: Factory function returning CliBackendProtocol
                 - None: No CLI backend
             **kwargs: Keyword arguments for core Agent
         """
-        # Resolve string CLI backends to instances in wrapper layer
-        if isinstance(cli_backend, str):
-            cli_backend = self._resolve_string_backend(cli_backend)
+        # Resolve CLI backend configuration using unified resolver
+        if cli_backend is not None and not self._is_protocol_instance(cli_backend):
+            from .cli_backends import resolve_cli_backend_config
+            cli_backend = resolve_cli_backend_config(cli_backend)
         
         # Pass resolved backend to core Agent
         super().__init__(*args, cli_backend=cli_backend, **kwargs)
     
-    def _resolve_string_backend(self, backend_id: str) -> Any:
-        """Resolve string backend ID to CliBackendProtocol instance.
-        
-        Args:
-            backend_id: Backend identifier (e.g., "claude-code")
-            
-        Returns:
-            CliBackendProtocol instance
-            
-        Raises:
-            ImportError: If CLI backends module not available
-            ValueError: If backend_id is not registered
-        """
-        try:
-            from .cli_backends import resolve_cli_backend
-            return resolve_cli_backend(backend_id)
-        except ImportError:
-            raise ImportError(
-                f"CLI backend '{backend_id}' requested but CLI backends not available. "
-                "This may indicate a packaging or installation issue."
-            )
+    def _is_protocol_instance(self, obj) -> bool:
+        """Check if object appears to be a protocol instance (duck typing)."""
+        return callable(obj) or hasattr(obj, 'process_turn')
 
 
 # Export the wrapper Agent as the default for import from praisonai

src/praisonai/praisonai/agents_generator.py

@@ -62,45 +62,6 @@ def __dir__():
 # Note: OTEL_SDK_DISABLED moved to CLI entry point per issue requirements
 
 
-def safe_format(template: str, **kwargs) -> str:
-    """
-    Safely format a string template, preserving JSON-like curly braces.
-    
-    This handles cases where templates contain Gutenberg block syntax like
-    {"level":2} which would cause KeyError with standard .format().
-    
-    Uses a two-pass approach:
-    1. Escape all {{ and }} (already escaped braces)
-    2. Only substitute known variable placeholders
-    
-    Args:
-        template: String template with {variable} placeholders
-        **kwargs: Variable substitutions to apply
-        
-    Returns:
-        Formatted string with variables substituted and JSON preserved
-        
-    Example:
-        >>> safe_format('Use <!-- wp:heading {"level":2} --> for {topic}', topic='AI')
-        'Use <!-- wp:heading {"level":2} --> for AI'
-    """
-    import re
-    
-    # Pattern to match {word} but not {"key": or {number} patterns
-    # This matches simple variable names like {topic}, {style}, etc.
-    def replace_var(match):
-        var_name = match.group(1)
-        if var_name in kwargs:
-            return str(kwargs[var_name])
-        # If not in kwargs, leave it as-is (don't raise KeyError)
-        return match.group(0)
-    
-    # Match {variable_name} where variable_name is a valid Python identifier
-    # but NOT {" (JSON start) or {number (like {2})
-    pattern = r'\{([a-zA-Z_][a-zA-Z0-9_]*)\}'
-    
-    return re.sub(pattern, replace_var, template)
-
 
 def _wrap_with_timeout(tool, timeout_seconds: float):
     """Enforce per-call timeout on a tool, sync or async, without
@@ -208,47 +169,10 @@ def sanitize_agent_name_for_autogen_v4(name):
 
 def _resolve_yaml_cli_backend(cli_backend_config, logger):
     """Resolve a YAML ``cli_backend`` field to a CliBackendProtocol instance.
-
-    Accepts ``None`` (no backend), a string id (e.g. ``"claude-code"``), or a
-    dict of shape ``{"id": "claude-code", "overrides": {...}}``. Returns
-    ``None`` on any error after logging a warning, so YAML parsing never raises.
-
-    Kept at module scope so it is unit-testable without constructing a full
-    ``AgentsGenerator`` instance.
+    Deprecated wrapper. Use praisonai.cli_backends.resolve_cli_backend_config directly.
     """
-    if not cli_backend_config:
-        return None
-
-    # Pre-seed label from config before import so we can show it in error logs
-    if isinstance(cli_backend_config, str):
-        label = cli_backend_config
-    elif isinstance(cli_backend_config, dict):
-        label = cli_backend_config.get('id') or "<missing>"
-    else:
-        label = type(cli_backend_config).__name__
-
-    try:
-        from praisonai.cli_backends import resolve_cli_backend
-        if isinstance(cli_backend_config, str):
-            return resolve_cli_backend(cli_backend_config)
-        if isinstance(cli_backend_config, dict):
-            backend_id = cli_backend_config.get('id')
-            if not backend_id:
-                raise ValueError("cli_backend dict must contain an 'id' field")
-            overrides = cli_backend_config.get('overrides') or {}
-            return resolve_cli_backend(backend_id, overrides=overrides)
-        raise ValueError(
-            f"cli_backend must be string or dict, got: {type(cli_backend_config).__name__}"
-        )
-    except ImportError:
-        logger.warning(
-            "CLI backend '%s' requested but not available", label
-        )
-    except Exception as e:
-        logger.warning(
-            "Failed to resolve CLI backend '%s': %s", label, e
-        )
-    return None
+    from praisonai.cli_backends import resolve_cli_backend_config
+    return resolve_cli_backend_config(cli_backend_config)
 
 
 class AgentsGenerator:
@@ -419,6 +343,27 @@ class AgentsGenerator:
                     agent_config[field] = value
                     self.logger.debug(f"CLI override for agent {agent_name}: {field} = {value}")
 
+    def _validate_cli_backend_compatibility(self, config, framework):
+        """Validate that cli_backend is only used with compatible frameworks."""
+        # Check if any agent/role defines cli_backend (support both key names)
+        all_entities = {
+            **config.get('roles', {}),
+            **config.get('agents', {}),
+        }
+        has_cli_backend = any(
+            isinstance(details, dict) and details.get('cli_backend')
+            for details in all_entities.values()
+        )
+        
+        if has_cli_backend and framework != 'praisonai':
+            self.logger.error(
+                f"cli_backend is not supported for framework='{framework}'. "
+                f"Remove cli_backend from your YAML or switch to framework='praisonai'."
+            )
+            raise ValueError(
+                f"cli_backend requires framework='praisonai', but framework='{framework}' was specified"
+            )
+
     def _validate_agents_config(self, config):
         """
         Validate agent configuration for typos in field names and provide suggestions.
@@ -610,18 +555,14 @@ class AgentsGenerator:
                             if isinstance(t, str) and t.strip():
                                 needed_tools.add(t.strip())
 
-                # Resolve only the tools actually referenced in YAML
+                # Resolve only the tools actually referenced in YAML using ToolResolver with instantiation
                 for tool_name in needed_tools:
                     try:
-                        resolved_tool = self.tool_resolver.resolve(tool_name)
-                        if resolved_tool is None:
-                            self.logger.warning(f"Tool '{tool_name}' not found")
-                            continue
-                        tools_dict[tool_name] = (
-                            resolved_tool() if inspect.isclass(resolved_tool) else resolved_tool
-                        )
+                        resolved_tool = self.tool_resolver.resolve(tool_name, instantiate=True)
+                        if resolved_tool is not None:
+                            tools_dict[tool_name] = resolved_tool
                     except Exception as e:
-                        self.logger.warning(f"Failed to initialize tool '{tool_name}': {e}")
+                        self.logger.warning(f"Failed to resolve or instantiate tool '{tool_name}': {e}")
                         continue
                             
             except Exception as e:
@@ -839,6 +780,9 @@ class AgentsGenerator:
         from .framework_adapters.validators import assert_framework_available
         assert_framework_available(framework)
         
+        # Validate cli_backend compatibility
+        self._validate_cli_backend_compatibility(config, framework)
+        
         self.logger.info(f"Using framework: {framework}")
         return await self.framework_adapter.arun(
             config,
@@ -851,10 +795,58 @@ class AgentsGenerator:
         )
 
     async def _arun_yaml_workflow(self, config):
-        """Async placeholder for workflow mode - currently delegates to sync version."""
-        # For now, delegate to sync version using asyncio.to_thread
-        import asyncio
-        return await asyncio.to_thread(self._run_yaml_workflow, config)
+        """
+        Async version of _run_yaml_workflow using YAMLWorkflowParser.
+        
+        This method handles agents.yaml files that have:
+        - process: workflow
+        
+        Args:
+            config: YAML configuration dictionary
+            
+        Returns:
+            str: Workflow execution result
+        """
+        if not is_available("praisonaiagents"):
+            raise ImportError("PraisonAI is not installed. Please install it with 'pip install praisonaiagents'")
+        
+        try:
+            from praisonaiagents.workflows import YAMLWorkflowParser
+        except ImportError as err:
+            raise ImportError("YAMLWorkflowParser not available. Please update praisonaiagents.") from err
+        
+        # Ensure name is present
+        if 'name' not in config:
+            config['name'] = config.get('topic', 'Workflow')
+        
+        # Pass model from config_list to workflow as default_llm
+        if self.config_list and self.config_list[0].get('model'):
+            model_from_cli = self.config_list[0]['model']
+            if 'workflow' not in config:
+                config['workflow'] = {}
+            if 'default_llm' not in config['workflow']:
+                config['workflow']['default_llm'] = model_from_cli
+        
+        import yaml as yaml_module
+        yaml_content = yaml_module.dump(config, default_flow_style=False)
+        
+        parser = YAMLWorkflowParser()
+        workflow = parser.parse_string(yaml_content)
+        
+        input_data = config.get('input', config.get('topic', ''))
+        
+        self.logger.info(f"Starting async YAML workflow with topic: {input_data}")
+        
+        if hasattr(workflow, 'astart'):
+            result = await workflow.astart(input_data)
+        else:
+            import asyncio
+            result = await asyncio.to_thread(workflow.start, input_data)
+            
+        if result.get("status") == "completed":
+            return result.get("output", "Workflow completed successfully")
+        else:
+            return f"Workflow failed: {result.get('error', 'Unknown error')}"
 
     def _run_yaml_workflow(self, config):
         """
@@ -911,3 +903,27 @@ class AgentsGenerator:
             return result.get("output", "Workflow completed successfully")
         else:
             return f"Workflow failed: {result.get('error', 'Unknown error')}"
+
+
+# Standalone function for backward compatibility with tests
+def safe_format(template, **kwargs):
+    """
+    Safe string formatting that preserves JSON/dict literals while substituting variables.
+    
+    This function only substitutes placeholders that look like identifiers (e.g., {topic})
+    while preserving JSON structures like {"level": 2}.
+    
+    Args:
+        template (str): Template string with placeholders
+        **kwargs: Values to substitute
+        
+    Returns:
+        str: Formatted string with safe substitutions
+    """
+    # Use the same regex-based substitution logic as BaseFrameworkAdapter._format_template
+    def replace_placeholder(match):
+        placeholder = match.group(1)
+        return str(kwargs.get(placeholder, match.group(0)))
+    
+    # Only replace placeholders that look like identifiers
+    return re.sub(r'\{([a-zA-Z_][a-zA-Z0-9_]*)\}', replace_placeholder, template)

src/praisonai/praisonai/bots/telegram.py

@@ -277,8 +277,11 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
         async def handle_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
             if not update.message or not update.message.text:
                 return
-            
-            message = self._convert_update_to_message(update)
+
+            message = await process_inbound_telegram_message(update, self)
+            if not message:
+                return
+
             command = message.command
             
             if command and command in self._command_handlers:
@@ -294,18 +297,25 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
         async def handle_status(update: Update, context: ContextTypes.DEFAULT_TYPE):
             if not update.message:
                 return
+            if not await process_inbound_telegram_message(update, self):
+                return
             await update.message.reply_text(self._format_status())
         
         async def handle_new(update: Update, context: ContextTypes.DEFAULT_TYPE):
             if not update.message:
                 return
-            user_id = str(update.message.from_user.id) if update.message.from_user else "unknown"
+            message = await process_inbound_telegram_message(update, self)
+            if not message:
+                return
+            user_id = message.sender.user_id if message.sender else "unknown"
             self._session.reset(user_id)
             await update.message.reply_text("Session reset. Starting fresh conversation.")
         
         async def handle_help(update: Update, context: ContextTypes.DEFAULT_TYPE):
             if not update.message:
                 return
+            if not await process_inbound_telegram_message(update, self):
+                return
             await update.message.reply_text(self._format_help())
         
         self._application.add_handler(CommandHandler("status", handle_status))

src/praisonai/praisonai/cli/features/bots_cli.py

@@ -1019,25 +1019,13 @@ class BotHandler:
         return tools
     
     def _resolve_tool_by_name(self, name: str):
-        """Resolve a tool by name."""
-        try:
-            import praisonai_tools
-            tool_class = getattr(praisonai_tools, name, None)
-            if tool_class:
-                return tool_class()
-        except ImportError:
-            pass
-        
-        try:
-            from praisonaiagents import tools as agent_tools
-            tool_func = getattr(agent_tools, name, None)
-            if tool_func:
-                return tool_func
-        except ImportError:
-            pass
-        
-        logger.warning(f"Tool not found: {name}")
-        return None
+        """Resolve a tool by name using the canonical ToolResolver."""
+        from praisonai.tool_resolver import ToolResolver
+        resolver = ToolResolver()
+        tool = resolver.resolve(name, instantiate=True)
+        if tool is None:
+            logger.warning(f"Tool not found: {name}")
+        return tool
     
     def _resolve_tools(self, tool_names: List[str]) -> List:
         """Resolve a list of tool names to tool instances."""

src/praisonai/praisonai/cli/features/job_workflow.py

@@ -667,7 +667,7 @@ class JobWorkflowExecutor:
 
     def _resolve_agent_tools(self, tool_names: List) -> List:
         """
-        Resolve tool names to actual tool functions.
+        Resolve tool names to actual tool functions using the canonical ToolResolver.
         
         Args:
             tool_names: List of tool names (strings) or tool functions
@@ -678,38 +678,19 @@ class JobWorkflowExecutor:
         if not tool_names:
             return []
         
+        from praisonai.tool_resolver import ToolResolver
+        resolver = ToolResolver()
+        
         resolved = []
         for tool in tool_names:
             if callable(tool):
                 # Already a function
                 resolved.append(tool)
             elif isinstance(tool, str):
-                # Try to resolve from registry
-                try:
-                    from praisonaiagents.tools import get_tool
-                    resolved_tool = get_tool(tool)
-                    if resolved_tool:
-                        resolved.append(resolved_tool)
-                except (ImportError, Exception):
-                    # Try common tools
-                    if tool == "execute_command":
-                        try:
-                            from praisonaiagents.tools import execute_command
-                            resolved.append(execute_command)
-                        except ImportError:
-                            pass
-                    elif tool == "read_file":
-                        try:
-                            from praisonaiagents.tools import read_file
-                            resolved.append(read_file)
-                        except ImportError:
-                            pass
-                    elif tool == "write_file":
-                        try:
-                            from praisonaiagents.tools import write_file
-                            resolved.append(write_file)
-                        except ImportError:
-                            pass
+                # Use canonical ToolResolver
+                resolved_tool = resolver.resolve(tool, instantiate=True)
+                if resolved_tool:
+                    resolved.append(resolved_tool)
         return resolved
 
     # ------------------------------------------------------------------

src/praisonai/praisonai/cli/session/unified.py

@@ -8,15 +8,25 @@ Uses JSON-based persistence with file locking for multi-process safety.
 import json
 import logging
 import os
+import sys
 import uuid
-import fcntl
 from dataclasses import dataclass, field, asdict
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+# fcntl is Unix-only; on Windows, use msvcrt for file locking
+try:
+    import fcntl
+    _HAS_FCNTL = True
+except ImportError:
+    _HAS_FCNTL = False
+
 logger = logging.getLogger(__name__)
 
+# Module-level sentinel to track if we've warned about degraded locking
+_WARNED_NO_FCNTL = False
+
 # Default session directory
 DEFAULT_SESSION_DIR = Path.home() / ".praison" / "sessions"
 
@@ -118,7 +128,7 @@ class UnifiedSessionStore:
         Args:
             session_dir: Directory to store sessions. Defaults to ~/.praison/sessions/
         """
-        self.session_dir = session_dir or DEFAULT_SESSION_DIR
+        self.session_dir = Path(session_dir) if session_dir else DEFAULT_SESSION_DIR
         self.session_dir.mkdir(parents=True, exist_ok=True)
         self._cache: Dict[str, UnifiedSession] = {}
         self._last_session_id: Optional[str] = None
@@ -142,12 +152,53 @@ class UnifiedSessionStore:
         session.updated_at = datetime.now().isoformat()
         
         try:
-            with open(path, 'w') as f:
-                fcntl.flock(f.fileno(), fcntl.LOCK_EX)
-                try:
-                    json.dump(session.to_dict(), f, indent=2)
-                finally:
-                    fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+            # Open in r+b mode to avoid truncation before locking
+            # Create file if it doesn't exist
+            if not path.exists():
+                path.touch()
+            
+            with open(path, 'r+b') as f:
+                # Cross-platform file locking
+                if sys.platform == "win32":
+                    # Windows locking - ensure consistent file position
+                    import msvcrt
+                    f.seek(0)
+                    msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, 1)  # Use blocking lock
+                    try:
+                        f.seek(0)
+                        f.truncate()  # Clear file after acquiring lock
+                        json_data = json.dumps(session.to_dict(), indent=2).encode('utf-8')
+                        f.write(json_data)
+                        f.flush()
+                        os.fsync(f.fileno())  # Force data to disk before unlock
+                    finally:
+                        f.seek(0)  # Return to start position for unlock
+                        msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
+                elif _HAS_FCNTL:
+                    # Unix locking
+                    fcntl.flock(f.fileno(), fcntl.LOCK_EX)
+                    try:
+                        f.seek(0)
+                        f.truncate()  # Clear file after acquiring lock
+                        json_data = json.dumps(session.to_dict(), indent=2).encode('utf-8')
+                        f.write(json_data)
+                        f.flush()
+                        os.fsync(f.fileno())  # Force data to disk before unlock
+                    finally:
+                        fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+                else:
+                    # 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 (fcntl not available); "
+                            "concurrent writers may corrupt session files."
+                        )
+                        _WARNED_NO_FCNTL = True
+                    f.seek(0)
+                    f.truncate()
+                    json_data = json.dumps(session.to_dict(), indent=2).encode('utf-8')
+                    f.write(json_data)
             
             # Update cache
             self._cache[session.session_id] = session
@@ -179,12 +230,31 @@ class UnifiedSessionStore:
             return None
         
         try:
-            with open(path, 'r') as f:
-                fcntl.flock(f.fileno(), fcntl.LOCK_SH)
-                try:
-                    data = json.load(f)
-                finally:
-                    fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+            with open(path, 'rb') as f:
+                # Cross-platform file locking (shared lock for reading)
+                if sys.platform == "win32":
+                    # Windows shared locking (read-only) - use blocking read lock
+                    import msvcrt
+                    f.seek(0)
+                    msvcrt.locking(f.fileno(), msvcrt.LK_RLCK, 1)  # Use shared/read lock
+                    try:
+                        json_data = f.read().decode('utf-8')
+                        data = json.loads(json_data)
+                    finally:
+                        f.seek(0)  # Return to start position for unlock
+                        msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, 1)
+                elif _HAS_FCNTL:
+                    # Unix shared locking
+                    fcntl.flock(f.fileno(), fcntl.LOCK_SH)
+                    try:
+                        json_data = f.read().decode('utf-8')
+                        data = json.loads(json_data)
+                    finally:
+                        fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+                else:
+                    # No locking available - just read
+                    json_data = f.read().decode('utf-8')
+                    data = json.loads(json_data)
             
             session = UnifiedSession.from_dict(data)
             self._cache[session_id] = session

src/praisonai/praisonai/cli_backends/__init__.py

@@ -6,6 +6,47 @@ Wrapper implementations of CLI backend protocols following AGENTS.md:
 - Lazy loading to avoid heavy imports at startup
 """
 
+def resolve_cli_backend_config(value):
+    """Resolve a YAML/Python cli_backend value (str | dict | instance) to a CliBackendProtocol.
+    
+    Unified resolver that handles both YAML and Python entry points consistently.
+    
+    Args:
+        value: CLI backend configuration - str, dict, or instance
+        
+    Returns:
+        CliBackendProtocol instance or None
+        
+    Raises:
+        ValueError: If configuration is invalid
+    """
+    if value is None:
+        return None
+        
+    # If already an instance, return as-is (duck typing)
+    if callable(value) or hasattr(value, 'process_turn'):
+        return value
+        
+    # Import registry functions
+    from .registry import resolve_cli_backend
+    
+    if isinstance(value, str):
+        if not value.strip():
+            raise ValueError("cli_backend string cannot be empty")
+        return resolve_cli_backend(value)
+    elif isinstance(value, dict):
+        backend_id = value.get('id')
+        if not backend_id:
+            raise ValueError("cli_backend dict must contain an 'id' field")
+        overrides = value.get('overrides') or {}
+        if not isinstance(overrides, dict):
+            raise ValueError("cli_backend.overrides must be a dict")
+        return resolve_cli_backend(backend_id, overrides=overrides)
+    else:
+        raise ValueError(
+            f"cli_backend must be string, dict, or instance, got: {type(value).__name__}"
+        )
+
 def __getattr__(name: str):
     """Lazy loading for CLI backend implementations."""
     if name == "register_cli_backend":
@@ -26,6 +67,7 @@ def __getattr__(name: str):
 __all__ = [
     "register_cli_backend",
     "resolve_cli_backend", 
+    "resolve_cli_backend_config",
     "ClaudeCodeBackend",
     "list_cli_backends"
 ]

src/praisonai/praisonai/framework_adapters/base.py

@@ -92,6 +92,34 @@ class FrameworkAdapter(Protocol):
         """
         ...
     
+    async def arun(
+        self,
+        config: Dict[str, Any],
+        llm_config: List[Dict],
+        topic: str,
+        *,
+        tools_dict: Optional[Dict[str, Any]] = None,
+        agent_callback: Optional[Callable] = None,
+        task_callback: Optional[Callable] = None,
+        cli_config: Optional[Dict[str, Any]] = None,
+    ) -> str:
+        """
+        Run the framework asynchronously with given configuration.
+        
+        Args:
+            config: Framework configuration
+            llm_config: LLM configuration list
+            topic: Topic for the tasks
+            tools_dict: Available tools dictionary
+            agent_callback: Callback for agent events
+            task_callback: Callback for task events
+            cli_config: CLI configuration
+            
+        Returns:
+            Execution result as string
+        """
+        ...
+    
     def cleanup(self) -> None:
         """Clean up any resources after execution."""
         ...
@@ -116,20 +144,41 @@ class BaseFrameworkAdapter:
         return list(self._tool_registry.keys())
     
     def _format_template(self, template: str, **kwargs) -> str:
-        """Safely format template string with given kwargs."""
-        try:
-            return template.format(**kwargs)
-        except KeyError as e:
-            # Import logger here to avoid circular imports
-            import logging
-            logger = logging.getLogger(__name__)
-            logger.warning("Template formatting failed for key %s; returning original template", e)
-            return template
-        except Exception as e:
-            import logging
-            logger = logging.getLogger(__name__)
-            logger.warning("Template formatting error: %s; returning original template", e)
+        """Safely format template string with given kwargs, preserving JSON-like braces."""
+        if not isinstance(template, str):
             return template
+        
+        import re
+        
+        def _sub(m):
+            name = m.group(1)
+            return str(kwargs[name]) if name in kwargs else m.group(0)
+        
+        # Only substitute simple variable names like {topic}, not JSON like {"level":2}
+        return re.sub(r'\{([a-zA-Z_][a-zA-Z0-9_]*)\}', _sub, template)
+    
+    async def arun(
+        self,
+        config: Dict[str, Any],
+        llm_config: List[Dict],
+        topic: str,
+        *,
+        tools_dict: Optional[Dict[str, Any]] = None,
+        agent_callback: Optional[Callable] = None,
+        task_callback: Optional[Callable] = None,
+        cli_config: Optional[Dict[str, Any]] = None,
+    ) -> str:
+        """
+        Default async implementation that falls back to thread-offloaded sync.
+        
+        Framework adapters with native async support should override this method.
+        """
+        import asyncio
+        return await asyncio.to_thread(
+            self.run, config, llm_config, topic,
+            tools_dict=tools_dict, agent_callback=agent_callback,
+            task_callback=task_callback, cli_config=cli_config
+        )
     
     def resolve(self) -> "FrameworkAdapter":
         """Default implementation returns self."""

src/praisonai/praisonai/framework_adapters/praisonai_adapter.py

@@ -23,6 +23,21 @@ class PraisonAIAdapter(BaseFrameworkAdapter):
         from .._framework_availability import is_available
         return is_available("praisonaiagents")
     
+    def _resolve_agent_model(self, details: Dict[str, Any], default_model: str) -> str:
+        """Resolve the LLM model for a specific agent, supporting per-agent configuration."""
+        llm_spec = details.get('llm')
+        
+        # Handle string format: llm: "gpt-4o-mini"
+        if isinstance(llm_spec, str) and llm_spec.strip():
+            return llm_spec.strip()
+        
+        # Handle dict format: llm: {"model": "groq/llama3-70b-8192"}
+        if isinstance(llm_spec, dict) and llm_spec.get('model'):
+            return llm_spec['model']
+        
+        # Fall back to global default
+        return default_model
+    
     def run(
         self,
         config: Dict[str, Any],
@@ -121,6 +136,9 @@ class PraisonAIAdapter(BaseFrameworkAdapter):
                     agent_tools = details.get('tools', [])
                     agent_tool_list = [tools_dict[t] for t in agent_tools if t in tools_dict]
                 
+                # Resolve per-agent LLM model
+                agent_model = self._resolve_agent_model(details, model_name)
+                
                 # Create basic agent
                 agent = PraisonAgent(
                     name=role_filled,
@@ -128,7 +146,7 @@ class PraisonAIAdapter(BaseFrameworkAdapter):
                     goal=goal_filled,
                     backstory=backstory_filled,
                     instructions=details.get('instructions'),
-                    llm=model_name,
+                    llm=agent_model,
                     allow_delegation=details.get('allow_delegation', False),
                     tools=agent_tool_list,
                 )
@@ -175,11 +193,13 @@ class PraisonAIAdapter(BaseFrameworkAdapter):
             memory = config.get('memory', False)
             
             if config.get('process') == 'hierarchical':
+                # Use specific manager_llm or fall back to global model
+                manager_model = config.get('manager_llm') or model_name
                 team = AgentTeam(
                     agents=list(agents.values()),
                     tasks=tasks,
                     process="hierarchical",
-                    manager_llm=config.get('manager_llm') or model_name,
+                    manager_llm=manager_model,
                     memory=memory
                 )
             else:
@@ -212,3 +232,182 @@ class PraisonAIAdapter(BaseFrameworkAdapter):
                 except Exception as e:
                     logger.error(f"Error stopping InteractiveRuntime: {e}")
     
+    async def arun(
+        self,
+        config: Dict[str, Any],
+        llm_config: List[Dict],
+        topic: str,
+        *,
+        tools_dict: Optional[Dict[str, Any]] = None,
+        agent_callback = None,
+        task_callback = None,
+        cli_config: Optional[Dict[str, Any]] = None,
+    ) -> str:
+        """
+        Run PraisonAI agents asynchronously using the native async path.
+        
+        This uses AgentTeam.astart() instead of thread offloading for true async execution.
+        """
+        # Import PraisonAI components only when needed
+        from praisonaiagents import Agent as PraisonAgent, Task as PraisonTask, AgentTeam
+        from .._framework_availability import is_available
+        import os
+        
+        logger.info("Starting PraisonAI async execution...")
+        
+        agents = {}
+        tasks = []
+        
+        # Get model from llm_config or environment
+        model_name = "gpt-4o-mini"
+        if llm_config and llm_config[0].get('model'):
+            model_name = llm_config[0]['model']
+        
+        # Initialize InteractiveRuntime for ACP/LSP if enabled globally
+        global_config = config.get('config', {})
+        acp_enabled = global_config.get('acp', False)
+        lsp_enabled = global_config.get('lsp', False)
+        interactive_runtime = None
+        
+        if acp_enabled or lsp_enabled:
+            try:
+                from praisonai.cli.features.interactive_runtime import InteractiveRuntime, RuntimeConfig
+                from praisonai.cli.features.agent_tools import create_agent_centric_tools
+                
+                # Use scoped configuration instead of process-global mutations
+                runtime_config = RuntimeConfig(
+                    workspace=os.getcwd(),
+                    acp_enabled=acp_enabled,
+                    lsp_enabled=lsp_enabled,
+                    approval_mode=os.environ.get("PRAISONAI_APPROVAL_MODE", "prompt")
+                )
+                rt = InteractiveRuntime(runtime_config)
+                logger.info(f"Starting InteractiveRuntime (ACP: {acp_enabled}, LSP: {lsp_enabled})")
+                
+                # Start the runtime asynchronously
+                await rt.start()
+                interactive_runtime = rt  # only assign AFTER start() succeeds
+                
+            except ImportError as e:
+                logger.warning(f"InteractiveRuntime not available: {e}")
+                interactive_runtime = None
+            except (RuntimeError, OSError, ConnectionError) as e:
+                logger.warning(f"InteractiveRuntime startup failed: {e}")
+                interactive_runtime = None
+        
+        try:
+            # All work that can throw *after* start() lives here
+            if interactive_runtime is not None:
+                from praisonai.cli.features.agent_tools import create_agent_centric_tools
+                centric_tools = create_agent_centric_tools(interactive_runtime)
+                logger.info(f"Loaded {len(centric_tools)} InteractiveRuntime tools")
+                tools_dict = {**(tools_dict or {}), **centric_tools}
+
+            # Create agents from roles - same logic as sync version
+            for role, details in config.get('roles', {}).items():
+                role_filled = self._format_template(details.get('role', role), topic=topic)
+                goal_filled = self._format_template(details.get('goal', ''), topic=topic)
+                backstory_filled = self._format_template(details.get('backstory', ''), topic=topic)
+                
+                # Resolve tools for this agent from tools_dict
+                agent_tool_list = []
+                if tools_dict:
+                    agent_tools = details.get('tools', [])
+                    agent_tool_list = [tools_dict[t] for t in agent_tools if t in tools_dict]
+                
+                # Resolve per-agent LLM model
+                agent_model = self._resolve_agent_model(details, model_name)
+                
+                # Create basic agent
+                agent = PraisonAgent(
+                    name=role_filled,
+                    role=role_filled,
+                    goal=goal_filled,
+                    backstory=backstory_filled,
+                    instructions=details.get('instructions'),
+                    llm=agent_model,
+                    allow_delegation=details.get('allow_delegation', False),
+                    tools=agent_tool_list,
+                )
+                
+                if agent_callback:
+                    agent.step_callback = agent_callback
+                    
+                agents[role] = agent
+                
+                # Create tasks for the agent - same logic as sync version
+                agent_tasks = details.get('tasks', {})
+                if not agent_tasks:
+                    # Auto-generate a task
+                    task_description = details.get('instructions') or backstory_filled
+                    task = PraisonTask(
+                        description=task_description,
+                        expected_output="Complete the assigned task successfully.",
+                        agent=agent,
+                    )
+                    if task_callback:
+                        task.callback = task_callback
+                    tasks.append(task)
+                else:
+                    for task_name, task_details in agent_tasks.items():
+                        description_filled = self._format_template(
+                            task_details['description'], topic=topic
+                        )
+                        expected_output_filled = self._format_template(
+                            task_details['expected_output'], topic=topic
+                        )
+                        
+                        task = PraisonTask(
+                            description=description_filled,
+                            expected_output=expected_output_filled,
+                            agent=agent,
+                        )
+                        
+                        if task_callback:
+                            task.callback = task_callback
+                        
+                        tasks.append(task)
+            
+            # Create and run the team asynchronously
+            memory = config.get('memory', False)
+            
+            if config.get('process') == 'hierarchical':
+                # Use specific manager_llm or fall back to global model
+                manager_model = config.get('manager_llm') or model_name
+                team = AgentTeam(
+                    agents=list(agents.values()),
+                    tasks=tasks,
+                    process="hierarchical",
+                    manager_llm=manager_model,
+                    memory=memory
+                )
+            else:
+                team = AgentTeam(
+                    agents=list(agents.values()),
+                    tasks=tasks,
+                    memory=memory
+                )
+            
+            # Use native async path instead of team.start()
+            response = await team.astart()
+            result = f"### PraisonAI Output ###\n{response}" if response else "### PraisonAI Output ###\nTask completed."
+            
+            # AgentOps integration if available
+            if is_available("agentops"):
+                import agentops
+                try:
+                    agentops.end_session("Success")
+                except Exception as e:  # noqa: BLE001 -- agentops errors must not crash the caller
+                    logger.warning(f"agentops.end_session failed: {e}")
+            
+            logger.info("PraisonAI async execution completed")
+            return result
+        finally:
+            # Cleanup InteractiveRuntime if it was started
+            if interactive_runtime is not None:
+                try:
+                    logger.info("Stopping InteractiveRuntime")
+                    await interactive_runtime.stop()
+                except Exception as e:
+                    logger.error(f"Error stopping InteractiveRuntime: {e}")
+    

src/praisonai/praisonai/gateway/server.py

@@ -1959,18 +1959,28 @@ class WebSocketGateway:
         async def handle_status(update: Update, context: Any):
             if not update.message:
                 return
+            from praisonai.bots.telegram import process_inbound_telegram_message
+            if not await process_inbound_telegram_message(update, bot):
+                return
             await update.message.reply_text(bot._format_status())
 
         async def handle_new(update: Update, context: Any):
             if not update.message:
                 return
-            user_id = str(update.message.from_user.id) if update.message.from_user else "unknown"
+            from praisonai.bots.telegram import process_inbound_telegram_message
+            message = await process_inbound_telegram_message(update, bot)
+            if not message:
+                return
+            user_id = message.sender.user_id if message.sender else "unknown"
             bot._session.reset(user_id)
             await update.message.reply_text("Session reset. Starting fresh conversation.")
 
         async def handle_help(update: Update, context: Any):
             if not update.message:
                 return
+            from praisonai.bots.telegram import process_inbound_telegram_message
+            if not await process_inbound_telegram_message(update, bot):
+                return
             await update.message.reply_text(bot._format_help())
 
         # Register handlers

src/praisonai/praisonai/tool_resolver.py

@@ -107,8 +107,7 @@ class ToolResolver:
                 cache: Dict[str, Callable] = {}
                 for name, obj in inspect.getmembers(module):
                     if (not name.startswith('_') and 
-                        callable(obj) and 
-                        not inspect.isclass(obj)):
+                        callable(obj)):
                         cache[name] = obj
                         logger.debug(f"Loaded local tool: {name}")
                 
@@ -246,7 +245,7 @@ class ToolResolver:
         
         return None
     
-    def resolve(self, name: str) -> Optional[Callable]:
+    def resolve(self, name: str, instantiate: bool = False) -> Optional[Callable]:
         """Resolve a tool name to a callable.
         
         Resolution order:
@@ -258,6 +257,7 @@ class ToolResolver:
         
         Args:
             name: Tool name to resolve
+            instantiate: If True, instantiate class tools automatically
             
         Returns:
             Callable if found, None if not found
@@ -281,6 +281,9 @@ class ToolResolver:
             # Double-check inside lock
             cached = self._resolve_cache.get(name, _SENTINEL)
             if cached is not _SENTINEL:
+                # Apply instantiation to cached result if needed
+                if instantiate and self._is_class(cached):
+                    return cached()
                 return cached
 
             # 1. Check local tools.py first (highest priority)
@@ -288,30 +291,40 @@ class ToolResolver:
                 logger.debug(f"Resolved '{name}' from local tools.py")
                 tool = local_tools[name]
                 self._resolve_cache[name] = tool
+                if instantiate and self._is_class(tool):
+                    return tool()
                 return tool
             
             # 2. Check wrapper ToolRegistry (NEW - ahead of SDK paths)
             tool = self._resolve_from_wrapper_registry(name)
             if tool is not None:
                 self._resolve_cache[name] = tool
+                if instantiate and self._is_class(tool):
+                    return tool()
                 return tool
             
             # 3. Check praisonaiagents.tools
             tool = self._resolve_from_praisonaiagents(name)
             if tool is not None:
                 self._resolve_cache[name] = tool
+                if instantiate and self._is_class(tool):
+                    return tool()
                 return tool
             
             # 4. Check praisonai-tools package
             tool = self._resolve_from_praisonai_tools(name)
             if tool is not None:
                 self._resolve_cache[name] = tool
+                if instantiate and self._is_class(tool):
+                    return tool()
                 return tool
             
             # 5. Check core SDK tool registry
             tool = self._resolve_from_registry(name)
             if tool is not None:
                 self._resolve_cache[name] = tool
+                if instantiate and self._is_class(tool):
+                    return tool()
                 return tool
             
             # Cache the None result to avoid repeated failed lookups
@@ -319,6 +332,11 @@ class ToolResolver:
             self._resolve_cache[name] = None
             return None
     
+    def _is_class(self, obj) -> bool:
+        """Check if object is a class (not an instance)."""
+        import inspect
+        return inspect.isclass(obj)
+    
     def resolve_many(self, names: List[str]) -> List[Callable]:
         """Resolve multiple tool names to callables.
         

src/praisonai/tests/unit/gateway/test_telegram_security_pipeline.py

@@ -65,8 +65,12 @@ def create_test_bot(allowed_users=None, allowed_channels=None, group_policy="men
         is_bot=True,
     )
     
-    # Mock the fire_message_received method
+    # Mock the fire_message_received method and other required attributes
     bot.fire_message_received = MagicMock()
+    bot._started_at = 1234567890.0
+    bot._agent = MagicMock()
+    bot._command_handlers = {}
+    bot._session = MagicMock()
     
     return bot
 
@@ -239,6 +243,85 @@ def test_security_pipeline_exists():
     assert callable(process_inbound_telegram_message), "Security pipeline function should be callable"
 
 
+@pytest.mark.asyncio
+@patch.object(UnknownUserHandler, 'handle')
+async def test_command_handlers_respect_user_allowlist(mock_unknown_handler):
+    """Built-in commands must pass the same security pipeline as text messages."""
+    mock_unknown_handler.return_value = False
+
+    bot = create_test_bot(allowed_users=["42"])
+    
+    # Mock the reply_text method to track if command handlers were called
+    reply_mock = AsyncMock()
+    
+    # Test that disallowed users are blocked by command handlers
+    for command in ("help", "status", "new"):
+        update = create_mock_telegram_update(
+            user_id="99",
+            text=f"/{command}",
+            chat_type="private",
+        )
+        update.message.reply_text = reply_mock
+        reply_mock.reset_mock()
+        
+        # Get the registered handler for this command from the bot's handlers
+        # We need to simulate how the telegram bot framework would call the handler
+        if command == "help":
+            from praisonai.bots.telegram import TelegramBot
+            # Create a handler like the bot does
+            async def test_handle_help(update, context):
+                if not update.message:
+                    return
+                if not await process_inbound_telegram_message(update, bot):
+                    return
+                await update.message.reply_text(bot._format_help())
+            await test_handle_help(update, None)
+        elif command == "status":
+            async def test_handle_status(update, context):
+                if not update.message:
+                    return
+                if not await process_inbound_telegram_message(update, bot):
+                    return
+                await update.message.reply_text(bot._format_status())
+            await test_handle_status(update, None)
+        elif command == "new":
+            async def test_handle_new(update, context):
+                if not update.message:
+                    return
+                message = await process_inbound_telegram_message(update, bot)
+                if not message:
+                    return
+                user_id = message.sender.user_id if message.sender else "unknown"
+                bot._session.reset(user_id)
+                await update.message.reply_text("Session reset. Starting fresh conversation.")
+            # Mock session reset
+            bot._session = MagicMock()
+            bot._session.reset = MagicMock()
+            await test_handle_new(update, None)
+        
+        # Assert the command handler did not reply (because security blocked it)
+        reply_mock.assert_not_called(), f"/{command} from disallowed user should not reply"
+
+    # Test that allowed users can use commands
+    allowed_update = create_mock_telegram_update(
+        user_id="42",
+        text="/help",
+        chat_type="private",
+    )
+    allowed_update.message.reply_text = reply_mock
+    reply_mock.reset_mock()
+    
+    async def test_handle_help_allowed(update, context):
+        if not update.message:
+            return
+        if not await process_inbound_telegram_message(update, bot):
+            return
+        await update.message.reply_text(bot._format_help())
+    
+    await test_handle_help_allowed(allowed_update, None)
+    reply_mock.assert_called_once(), "Commands from allowed users should reply"
+
+
 @pytest.mark.asyncio
 async def test_shared_pipeline_consistency():
     """Test that the shared pipeline provides consistent results."""

tests/test_wrapper_layer_regression.py

@@ -0,0 +1,213 @@
+"""
+Regression tests for the three wrapper layer gaps fixed in Issue #1646.
+
+Tests ensure:
+1. InteractiveRuntime lifecycle is properly managed
+2. Tool resolution is consistent across all entry points
+3. CLI backend validation works correctly across frameworks
+"""
+
+import pytest
+from unittest.mock import MagicMock, patch
+
+
+class TestInteractiveRuntimeLifecycle:
+    """Test InteractiveRuntime is not stopped before agents.start()"""
+    
+    def test_interactive_runtime_lifecycle_mock(self):
+        """Test that InteractiveRuntime.stop() is not called prematurely"""
+        with patch('praisonai.praisonai.agents_generator.asyncio') as mock_asyncio, \
+             patch('praisonai.praisonai.agents_generator.InteractiveRuntime') as MockRuntime, \
+             patch('praisonai.praisonai.agents_generator.create_agent_centric_tools') as mock_tools:
+            
+            # Setup mocks
+            mock_loop = MagicMock()
+            mock_asyncio.new_event_loop.return_value = mock_loop
+            mock_runtime = MockRuntime.return_value
+            mock_tools.return_value = []
+            
+            from praisonai.praisonai.agents_generator import AgentsGenerator
+            
+            generator = AgentsGenerator(
+                agent_file=None, 
+                framework='praisonai',
+                config_list=[{"model": "gpt-4o-mini"}]
+            )
+            
+            # Config with ACP enabled
+            config = {
+                'config': {'acp': True},
+                'roles': {
+                    'test_agent': {
+                        'role': 'Test Agent',
+                        'goal': 'Test',
+                        'backstory': 'Testing',
+                        'tasks': {
+                            'test_task': {
+                                'description': 'Test task',
+                                'expected_output': 'Done'
+                            }
+                        }
+                    }
+                }
+            }
+            
+            # Run the configuration
+            try:
+                result = generator.run_generation(config)
+            except Exception as e:
+                # Expected to fail due to missing framework adapter or mocked dependencies
+                # Lifecycle test only cares that runtime.start() was called without premature stop()
+                pass
+            
+            # Verify runtime.start() was called but stop() was NOT called during setup
+            mock_loop.run_until_complete.assert_any_call(mock_runtime.start())
+            
+            # Should NOT have been stopped during the setup phase
+            # Since we expect failure before reaching agents.run(), stop() should not be called at all
+            mock_runtime.stop.assert_not_called()
+
+
+class TestToolResolutionConsistency:
+    """Test that tool resolution is consistent across all entry points"""
+    
+    def test_tool_resolver_instantiate_parameter(self):
+        """Test ToolResolver.resolve() instantiate parameter works correctly"""
+        from praisonai.praisonai.tool_resolver import ToolResolver
+        
+        resolver = ToolResolver()
+        
+        # Test with a mock class tool
+        class MockTool:
+            def __init__(self):
+                self.instantiated = True
+        
+        with patch.object(resolver, '_resolve_from_praisonai_tools') as mock_resolve:
+            mock_resolve.return_value = MockTool
+            
+            # Without instantiate=True, should return class
+            result = resolver.resolve('mock_tool', instantiate=False)
+            assert result == MockTool
+            
+            # With instantiate=True, should return instance
+            result = resolver.resolve('mock_tool', instantiate=True)
+            assert hasattr(result, 'instantiated')
+            assert result.instantiated is True
+
+    def test_bots_cli_uses_tool_resolver(self):
+        """Test that bots_cli._resolve_tool_by_name uses ToolResolver"""
+        from praisonai.praisonai.cli.features.bots_cli import BotHandler
+        
+        bots = BotHandler()
+        
+        with patch('praisonai.tool_resolver.ToolResolver') as MockResolver:
+            mock_resolver = MockResolver.return_value
+            mock_resolver.resolve.return_value = MagicMock()
+            
+            result = bots._resolve_tool_by_name('test_tool')
+            
+            # Verify ToolResolver was used
+            MockResolver.assert_called_once()
+            mock_resolver.resolve.assert_called_once_with('test_tool', instantiate=True)
+
+    def test_job_workflow_uses_tool_resolver(self):
+        """Test that job_workflow uses ToolResolver for tool resolution"""
+        from praisonai.praisonai.cli.features.job_workflow import JobWorkflowExecutor
+        
+        executor = JobWorkflowExecutor({}, "dummy_path.yaml")
+        
+        with patch('praisonai.tool_resolver.ToolResolver') as MockResolver:
+            mock_resolver = MockResolver.return_value
+            mock_resolver.resolve.return_value = lambda: "resolved_tool"
+            
+            result = executor._resolve_agent_tools(['test_tool'])
+            
+            # Verify ToolResolver was used
+            MockResolver.assert_called_once()
+            mock_resolver.resolve.assert_called_once_with('test_tool', instantiate=True)
+            assert result == ["resolved_tool"]
+
+
+class TestCliBackendValidation:
+    """Test CLI backend validation across frameworks"""
+    
+    def test_cli_backend_config_resolver(self):
+        """Test the unified CLI backend config resolver"""
+        from praisonai.praisonai.cli_backends import resolve_cli_backend_config
+        
+        # Test string input
+        with patch('praisonai.praisonai.cli_backends.registry.resolve_cli_backend') as mock_resolve:
+            mock_resolve.return_value = MagicMock()
+            
+            result = resolve_cli_backend_config("claude-code")
+            mock_resolve.assert_called_once_with("claude-code")
+        
+        # Test dict input
+        with patch('praisonai.praisonai.cli_backends.registry.resolve_cli_backend') as mock_resolve:
+            mock_resolve.return_value = MagicMock()
+            
+            config = {"id": "claude-code", "overrides": {"timeout": 30}}
+            result = resolve_cli_backend_config(config)
+            mock_resolve.assert_called_once_with("claude-code", overrides={"timeout": 30})
+        
+        # Test invalid dict (missing id)
+        with pytest.raises(ValueError, match="must contain an 'id' field"):
+            resolve_cli_backend_config({"overrides": {"timeout": 30}})
+
+    def test_framework_cli_backend_validation(self):
+        """Test that CLI backend validation works for unsupported frameworks"""
+        from praisonai.praisonai.agents_generator import AgentsGenerator
+        
+        generator = AgentsGenerator(
+            agent_file=None,
+            framework='autogen',
+            config_list=[{"model": "gpt-4o-mini"}]
+        )
+        
+        # Config with cli_backend on unsupported framework
+        config = {
+            'roles': {
+                'test_agent': {
+                    'role': 'Test Agent',
+                    'goal': 'Test',
+                    'backstory': 'Testing',
+                    'cli_backend': 'claude-code'  # Should trigger validation error
+                }
+            }
+        }
+        
+        # Should raise ValueError for incompatible framework
+        with pytest.raises(ValueError, match="cli_backend requires framework='praisonai'"):
+            generator._validate_cli_backend_compatibility(config, 'autogen')
+
+    def test_agent_wrapper_cli_backend_resolution(self):
+        """Test that Agent wrapper resolves CLI backend configs correctly"""
+        from praisonai.praisonai.agent import Agent
+        
+        # Test that string cli_backend is resolved
+        with patch('praisonai.praisonai.cli_backends.resolve_cli_backend_config') as mock_resolve:
+            mock_resolve.return_value = MagicMock()
+            
+            agent = Agent(name="test", cli_backend="claude-code")
+            mock_resolve.assert_called_once_with("claude-code")
+        
+        # Test that dict cli_backend is resolved
+        with patch('praisonai.praisonai.cli_backends.resolve_cli_backend_config') as mock_resolve:
+            mock_resolve.return_value = MagicMock()
+            
+            config = {"id": "claude-code", "overrides": {"timeout": 30}}
+            agent = Agent(name="test", cli_backend=config)
+            mock_resolve.assert_called_once_with(config)
+        
+        # Test that protocol instances pass through unchanged
+        mock_instance = MagicMock()
+        mock_instance.process_turn = MagicMock()  # Duck typing for protocol
+        
+        with patch('praisonai.praisonai.cli_backends.resolve_cli_backend_config') as mock_resolve:
+            agent = Agent(name="test", cli_backend=mock_instance)
+            # Should not call resolver for already-resolved instances
+            mock_resolve.assert_not_called()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])