Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 141 additions & 0 deletions src/praisonai/praisonai/ui/_external_agents.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
"""Shared helpers for wiring external agents into any UI entry point.

Single source of truth for:
- Listing installed external agents (lazy, cached)
- Rendering Chainlit Switch widgets / aiui settings entries
- Building the tools list from enabled agents
"""

from functools import lru_cache
from typing import Any, Callable, Dict, List, Optional

# Map of UI toggle id β†’ (integration class path, pretty label)
EXTERNAL_AGENTS: Dict[str, Dict[str, str]] = {
"claude_enabled": {"module": "claude_code", "cls": "ClaudeCodeIntegration",
"label": "Claude Code (coding, file edits)", "cli": "claude"},
"gemini_enabled": {"module": "gemini_cli", "cls": "GeminiCLIIntegration",
"label": "Gemini CLI (analysis, search)", "cli": "gemini"},
"codex_enabled": {"module": "codex_cli", "cls": "CodexCLIIntegration",
"label": "Codex CLI (refactoring)", "cli": "codex"},
"cursor_enabled": {"module": "cursor_cli", "cls": "CursorCLIIntegration",
"label": "Cursor CLI (IDE tasks)", "cli": "cursor-agent"},
}


@lru_cache(maxsize=1)
def installed_external_agents() -> List[str]:
"""Return toggle ids of external agents whose CLI is on PATH."""
import shutil
return [toggle_id for toggle_id, meta in EXTERNAL_AGENTS.items()
if shutil.which(meta["cli"])]
Comment on lines +25 to +30
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 lru_cache on installed_external_agents is process-lifetime β€” consider documenting the restart requirement

The @lru_cache(maxsize=1) caches the PATH check for the entire lifetime of the process. If a user installs a new CLI tool (e.g., npm install -g @google/gemini-cli) while the UI server is running, the new agent will not appear without restarting the server. This can be confusing.

At a minimum, add a docstring note:

@lru_cache(maxsize=1)
def installed_external_agents() -> List[str]:
    """Return toggle ids of external agents whose CLI is on PATH.

    Result is cached for the lifetime of the process. Restart the
    UI server after installing a new external agent CLI for it to appear.
    """

Alternatively, cache with a TTL (e.g., using cachetools.TTLCache) or expose a installed_external_agents.cache_clear() call via a settings-refresh hook.



def external_agent_tools(settings: Dict[str, Any], workspace: str = ".") -> list:
"""Build tools list from settings dict of toggle_id β†’ bool."""
import importlib
tools = []
for toggle_id, enabled in settings.items():
if not enabled or toggle_id not in EXTERNAL_AGENTS:
continue
meta = EXTERNAL_AGENTS[toggle_id]
try:
mod = importlib.import_module(f"praisonai.integrations.{meta['module']}")
integration = getattr(mod, meta["cls"])(workspace=workspace)
if integration.is_available:
tools.append(integration.as_tool())
except (ImportError, AttributeError):
continue # Integration module/class not available
except Exception as e: # noqa: BLE001 β€” isolate faulty integrations
import logging
logging.getLogger(__name__).warning(
"Skipping external agent %s due to error: %s", toggle_id, e
)
continue
return tools
Comment on lines +33 to +54
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟑 Minor

Too-narrow except β€” non-import errors from integration init/as_tool() will crash tool assembly.

importlib.import_module raises ImportError, and getattr raises AttributeError. But cls(workspace=workspace) and integration.as_tool() can raise anything (TypeError for signature mismatches, OSError during availability probes, third-party errors, etc.). A single misbehaving integration will then abort the entire external_agent_tools call, dropping all enabled agents β€” not just the faulty one.

πŸ› οΈ Proposed fix
-        try:
-            mod = importlib.import_module(f"praisonai.integrations.{meta['module']}")
-            integration = getattr(mod, meta["cls"])(workspace=workspace)
-            if integration.is_available:
-                tools.append(integration.as_tool())
-        except (ImportError, AttributeError):
-            # Integration module not available, skip
-            continue
+        try:
+            mod = importlib.import_module(f"praisonai.integrations.{meta['module']}")
+            integration = getattr(mod, meta["cls"])(workspace=workspace)
+            if integration.is_available:
+                tools.append(integration.as_tool())
+        except (ImportError, AttributeError):
+            continue  # Integration module/class not available
+        except Exception as e:  # noqa: BLE001 β€” isolate faulty integrations
+            import logging
+            logging.getLogger(__name__).warning(
+                "Skipping external agent %s due to error: %s", toggle_id, e
+            )
+            continue
πŸ€– Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@src/praisonai/praisonai/ui/_external_agents.py` around lines 33 - 49, In
external_agent_tools, broaden the try/except around importing, instantiating and
calling the integration so that any Exception raised by
getattr(...)[meta["cls"]](workspace=...) or integration.as_tool() (not just
ImportError/AttributeError) is caught and the faulty integration is skipped;
specifically, keep the import and getattr but catch Exception (or add a second
except Exception) for failures during meta["cls"](...) and
integration.as_tool(), log or ignore the exception, and continue so that a
single misbehaving integration does not abort building the full tools list
(referencing external_agent_tools, EXTERNAL_AGENTS, meta['module'], meta['cls'],
integration.is_available, and integration.as_tool()).



def chainlit_switches(current_settings: Dict[str, bool]):
"""Return Chainlit Switch widgets for installed external agents only."""
from chainlit.input_widget import Switch
return [
Switch(id=toggle_id, label=EXTERNAL_AGENTS[toggle_id]["label"],
initial=current_settings.get(toggle_id, False))
for toggle_id in installed_external_agents()
]


def aiui_settings_entries() -> Dict[str, Any]:
"""Return aiui settings entries for installed external agents."""
settings = {}
for toggle_id in installed_external_agents():
meta = EXTERNAL_AGENTS[toggle_id]
settings[toggle_id] = {
"type": "checkbox",
"label": meta["label"],
"default": False
}
return settings


def _parse_setting_bool(value: Any) -> bool:
if isinstance(value, bool):
return value
if value is None:
return False
return str(value).strip().lower() in {"true", "1", "yes", "on"}


def load_external_agent_settings_from_chainlit(
load_setting_fn: Optional[Callable[[str], Any]] = None,
) -> Dict[str, bool]:
"""Load external agent settings from Chainlit session and persistent storage.

Args:
load_setting_fn: Optional callback used to load persisted settings by key.
If omitted, falls back to importing ``praisonai.ui.chat.load_setting``
for backward compatibility.
"""
import chainlit as cl
settings = {toggle_id: False for toggle_id in EXTERNAL_AGENTS}
loader = load_setting_fn

# Try to load from persistent storage (if load_setting is available)
if loader is None:
try:
# Backward-compatible fallback for callers that don't pass a loader
from praisonai.ui.chat import load_setting as loader # type: ignore
except ImportError:
loader = None

if loader is not None:
legacy_claude = _parse_setting_bool(loader("claude_code_enabled"))

# Load all current toggles from persistent storage first
for toggle_id in EXTERNAL_AGENTS:
persistent_value = loader(toggle_id)
if persistent_value: # non-empty string means explicitly stored
settings[toggle_id] = _parse_setting_bool(persistent_value)

# Apply legacy migration only where no explicit value was stored
if legacy_claude and not loader("claude_enabled"):
settings["claude_enabled"] = True

# Load from session (may override persistent settings)
# Check for legacy key in session
if _parse_setting_bool(cl.user_session.get("claude_code_enabled", False)):
settings["claude_enabled"] = True

# Load all current toggles from session
for toggle_id in EXTERNAL_AGENTS:
session_value = cl.user_session.get(toggle_id)
if session_value is not None:
settings[toggle_id] = _parse_setting_bool(session_value)

return settings


def save_external_agent_settings_to_chainlit(settings: Dict[str, bool]):
"""Save external agent settings to Chainlit session."""
import chainlit as cl
for toggle_id, enabled in settings.items():
cl.user_session.set(toggle_id, enabled)
43 changes: 40 additions & 3 deletions src/praisonai/praisonai/ui/agents.py
Original file line number Diff line number Diff line change
Expand Up @@ -426,6 +426,26 @@ def step_callback_sync(step_details):
except Exception as e:
logger.error(f"Error in step_callback_sync: {e}", exc_info=True)

# Get external agent tools from current settings
external_tools = []
try:
from praisonai.ui._external_agents import external_agent_tools, EXTERNAL_AGENTS
settings = cl.user_session.get("settings", {})
external_settings = {k: settings.get(k, False) for k in EXTERNAL_AGENTS}
workspace = os.environ.get("PRAISONAI_WORKSPACE", ".")
external_tools = external_agent_tools(external_settings, workspace=workspace)
except ImportError:
pass

# Get existing tools from details
existing_tools = details.get('tools', [])
if isinstance(existing_tools, str):
# If tools is a string, it might be a module reference - keep as is
all_tools = existing_tools
else:
# Combine existing and external tools
all_tools = (existing_tools or []) + external_tools

agent = Agent(
name=role_name,
role=role_filled,
Expand All @@ -439,7 +459,8 @@ def step_callback_sync(step_details):
max_execution_time=details.get('max_execution_time'),
cache=details.get('cache', True),
step_callback=step_callback_sync,
reflection=details.get('self_reflect', False)
reflection=details.get('self_reflect', False),
tools=all_tools if all_tools else None
)
agents_map[role] = agent

Expand Down Expand Up @@ -485,8 +506,10 @@ def step_callback_sync(step_details):
logger.warning(f"Tool '{tool_name}' not found. Skipping.")

# Set the agent's tools after collecting all tools
if role_tools:
agent.tools = role_tools
# Merge resolved YAML tools with external agent tools so both survive
merged_tools = role_tools + external_tools
Comment thread
greptile-apps[bot] marked this conversation as resolved.
if merged_tools:
agent.tools = merged_tools

for tname, tdetails in details.get('tasks', {}).items():
description_filled = tdetails['description'].format(topic=topic)
Expand Down Expand Up @@ -674,6 +697,18 @@ async def start_chat():
with open("agents.yaml", "w") as f:
f.write("# Add your custom agents here\n")

# Load external agent settings
try:
from praisonai.ui._external_agents import (
chainlit_switches,
load_external_agent_settings_from_chainlit,
Comment thread
greptile-apps[bot] marked this conversation as resolved.
)
external_settings = load_external_agent_settings_from_chainlit(load_setting)
except ImportError:
def chainlit_switches(_settings): # no-op fallback
return []
external_settings = {}
Comment on lines +700 to +710
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟑 Minor

Duplicate import of chainlit_switches already performed inside the try.

Line 699 imports chainlit_switches unconditionally, while line 701 imports load_external_agent_settings_from_chainlit inside try/except ImportError. If _external_agents is missing, the unconditional import on line 699 will raise and the except will never fire. Either wrap both imports in the same try, or move chainlit_switches inside it.

πŸ› οΈ Proposed fix
-        # Load external agent settings
-        from praisonai.ui._external_agents import chainlit_switches
-        try:
-            from praisonai.ui._external_agents import load_external_agent_settings_from_chainlit
+        # Load external agent settings
+        try:
+            from praisonai.ui._external_agents import (
+                chainlit_switches,
+                load_external_agent_settings_from_chainlit,
+            )
             external_settings = load_external_agent_settings_from_chainlit()
         except ImportError:
+            def chainlit_switches(_settings):  # no-op fallback
+                return []
             external_settings = {}
πŸ€– Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@src/praisonai/praisonai/ui/agents.py` around lines 698 - 704, The
unconditional import of chainlit_switches will raise before the try/except for
load_external_agent_settings_from_chainlit can catch missing-module errors; move
the import of chainlit_switches into the same try block (or wrap both imports
together) so both from praisonai.ui._external_agents imports (chainlit_switches
and load_external_agent_settings_from_chainlit) are attempted inside the try and
on ImportError you set external_settings = {} and handle the absence of
chainlit_switches accordingly.


settings = await cl.ChatSettings(
[
TextInput(id="Model", label="OpenAI - Model", initial=model_name),
Expand All @@ -685,6 +720,7 @@ async def start_chat():
values=["praisonai", "crewai", "autogen"],
initial_index=0,
),
*chainlit_switches(external_settings)
]
).send()
cl.user_session.set("settings", settings)
Expand Down Expand Up @@ -719,6 +755,7 @@ async def start_chat():
),
TextInput(id="agents", label="agents.yaml", initial=yaml_content, multiline=True),
TextInput(id="tools", label="tools.py", initial=tools_content, multiline=True),
*chainlit_switches(external_settings)
]
).send()
cl.user_session.set("settings", settings)
Expand Down
74 changes: 63 additions & 11 deletions src/praisonai/praisonai/ui/chat.py
Original file line number Diff line number Diff line change
Expand Up @@ -400,28 +400,40 @@ def auth_callback(username: str, password: str):
logger.warning(f"Login failed for user: {username}")
return None

def _get_or_create_agent(model_name: str, tools_enabled: bool = True):
def _get_or_create_agent(model_name: str, tools_enabled: bool = True, external_agents_settings: dict = None):
"""Get or create a reusable agent for the session."""
Agent = _get_praisonai_agent()
if Agent is None:
return None
if external_agents_settings is None:
external_agents_settings = {}

# Get cached agent from session
cached_agent = cl.user_session.get("_cached_agent")
cached_model = cl.user_session.get("_cached_agent_model")
cached_external = cl.user_session.get("_cached_agent_external", {})

# Reuse if model matches
if cached_agent is not None and cached_model == model_name:
# Reuse if model and external settings match
if (cached_agent is not None and cached_model == model_name
and cached_external == external_agents_settings):
return cached_agent

# Create new agent with interactive tools
_profile_start("create_agent")
tools = []
if tools_enabled:
tools = _get_interactive_tools()
tools = list(_get_interactive_tools()) # Copy to avoid mutating cache
# Add Tavily if available
if os.getenv("TAVILY_API_KEY"):
tools.append(tavily_web_search)
# Add external agent tools
from praisonai.ui._external_agents import external_agent_tools
tools.extend(
external_agent_tools(
external_agents_settings,
workspace=os.environ.get("PRAISONAI_WORKSPACE", "."),
)
)
Comment on lines +429 to +436
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Unguarded import of _external_agents inside a hot path

Every other call-site that imports from praisonai.ui._external_agents wraps the import in a try/except ImportError (see agents.py lines 431–437, code.py start(), etc.). Here, the import is bare β€” if the module fails to load (e.g., a circular import during startup, a missing transitive dependency), _get_or_create_agent raises an unhandled ImportError that bubbles up on every message, effectively making the chat UI unusable.

Apply the same defensive pattern used elsewhere:

        # Add external agent tools
        try:
            from praisonai.ui._external_agents import external_agent_tools
            tools.extend(
                external_agent_tools(
                    external_agents_settings,
                    workspace=os.environ.get("PRAISONAI_WORKSPACE", "."),
                )
            )
        except ImportError:
            pass


agent = Agent(
name="PraisonAI Assistant",
Expand All @@ -443,10 +455,27 @@ def _get_or_create_agent(model_name: str, tools_enabled: bool = True):
# Cache the agent
cl.user_session.set("_cached_agent", agent)
cl.user_session.set("_cached_agent_model", model_name)
cl.user_session.set("_cached_agent_external", external_agents_settings)
_profile_end("create_agent")

return agent


def _parse_setting_bool(value):
if isinstance(value, bool):
return value
if value is None:
return False
return str(value).strip().lower() in {"true", "1", "yes", "on"}


def _get_external_agent_settings_from_session() -> dict:
from praisonai.ui._external_agents import EXTERNAL_AGENTS
return {
toggle_id: _parse_setting_bool(cl.user_session.get(toggle_id, False))
for toggle_id in EXTERNAL_AGENTS
}

@cl.on_chat_start
async def start():
_profile_start("on_chat_start")
Expand All @@ -459,8 +488,12 @@ async def start():
cl.user_session.set("tools_enabled", tools_enabled)
logger.debug(f"Model name: {model_name}, Tools enabled: {tools_enabled}")

# Load external agent settings
from praisonai.ui._external_agents import load_external_agent_settings_from_chainlit, chainlit_switches
external_settings = load_external_agent_settings_from_chainlit(load_setting)

# Pre-create agent for faster first response
_get_or_create_agent(model_name, tools_enabled)
_get_or_create_agent(model_name, tools_enabled, external_settings)

settings = cl.ChatSettings(
[
Expand All @@ -474,7 +507,8 @@ async def start():
id="tools_enabled",
label="Enable Tools (ACP, LSP, Web Search)",
initial=tools_enabled
)
),
*chainlit_switches(external_settings)
]
)
cl.user_session.set("settings", settings)
Expand Down Expand Up @@ -508,14 +542,24 @@ async def setup_agent(settings):
cl.user_session.set("model_name", model_name)
cl.user_session.set("tools_enabled", tools_enabled)

# Invalidate cached agent if model changed
# Save external agent settings
from praisonai.ui._external_agents import save_external_agent_settings_to_chainlit, EXTERNAL_AGENTS
external_agent_settings = {k: settings.get(k, False) for k in EXTERNAL_AGENTS}
save_external_agent_settings_to_chainlit(external_agent_settings)

# Invalidate cached agent if model changed or external agents changed
cached_model = cl.user_session.get("_cached_agent_model")
if cached_model != model_name:
cached_external = cl.user_session.get("_cached_agent_external", {})
if cached_model != model_name or cached_external != external_agent_settings:
cl.user_session.set("_cached_agent", None)
cl.user_session.set("_cached_agent_model", None)
cl.user_session.set("_cached_agent_external", None)

Comment on lines +545 to 557
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Cache invalidation check doesn't actually compare external-agent settings.

The comment at line 525 claims the cache is invalidated "if model changed or external agents changed", but the condition at line 527 only compares cached_model != model_name. If the user toggles Claude/Gemini/Codex/Cursor without changing the model, the cached agent is reused and the updated tools list is never assembled β€” toggle changes won't take effect until the next model change or session.

πŸ› οΈ Proposed fix
-    # Invalidate cached agent if model changed or external agents changed
+    # Invalidate cached agent if model changed or external agents changed
     cached_model = cl.user_session.get("_cached_agent_model")
-    if cached_model != model_name:
+    cached_external = cl.user_session.get("_cached_agent_external", {})
+    if cached_model != model_name or cached_external != external_settings:
         cl.user_session.set("_cached_agent", None)
         cl.user_session.set("_cached_agent_model", None)
+        cl.user_session.set("_cached_agent_external", None)

You'll also need to mirror this in _get_or_create_agent (cache key must include the external settings), as code.py already does at lines 336-340.

πŸ€– Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@src/praisonai/praisonai/ui/chat.py` around lines 520 - 530, The cache
invalidation only checks cached_model != model_name, so include a comparison of
the current external agent settings (built from EXTERNAL_AGENTS as
external_settings via save_external_agent_settings_to_chainlit) with the cached
settings in the session and clear the cached agent when they differ; update the
block that reads cl.user_session.get("_cached_agent_model") to also read and
compare cl.user_session.get("_cached_agent_external_settings") (or similar key)
against the new external_settings and call cl.user_session.set("_cached_agent",
None) and set the updated model and external settings when invalidating. Also
mirror this change inside _get_or_create_agent so the cache key/lookup includes
the external settings (i.e., include external_settings in whatever key or
session fields you use to decide reuse) so toggling Claude/Gemini/Codex/Cursor
forces rebuilding tools even if model_name is unchanged.

save_setting("model_name", model_name)
save_setting("tools_enabled", str(tools_enabled).lower())
# Save external agent settings to persistent storage
for toggle_id, enabled in external_agent_settings.items():
save_setting(toggle_id, str(enabled).lower())

thread_id = cl.user_session.get("thread_id")
if thread_id:
Expand Down Expand Up @@ -572,7 +616,8 @@ async def main(message: cl.Message):
msg = cl.Message(content="")

# Try PraisonAI Agent first (faster, with tool reuse)
agent = _get_or_create_agent(model_name, tools_enabled) if tools_enabled else None
external_settings = _get_external_agent_settings_from_session()
agent = _get_or_create_agent(model_name, tools_enabled, external_settings) if tools_enabled else None

if agent is not None:
_profile_start("agent_response")
Expand Down Expand Up @@ -781,6 +826,11 @@ async def on_chat_resume(thread: ThreadDict):
tools_enabled = (load_setting("tools_enabled") or "true").lower() == "true"

logger.debug(f"Model name: {model_name}")

# Load external agent settings for resume
from praisonai.ui._external_agents import load_external_agent_settings_from_chainlit, chainlit_switches
external_settings = load_external_agent_settings_from_chainlit()

settings = cl.ChatSettings(
[
TextInput(
Expand All @@ -793,7 +843,8 @@ async def on_chat_resume(thread: ThreadDict):
id="tools_enabled",
label="Enable Tools (ACP, LSP, Web Search)",
initial=tools_enabled
)
),
*chainlit_switches(external_settings),
]
)
await settings.send()
Expand Down Expand Up @@ -828,7 +879,8 @@ async def on_chat_resume(thread: ThreadDict):
cl.user_session.set("message_history", message_history)

# Pre-create agent for faster first response
_get_or_create_agent(model_name, tools_enabled)
external_settings = _get_external_agent_settings_from_session()
_get_or_create_agent(model_name, tools_enabled, external_settings)

image_data = metadata.get("image")
if image_data:
Expand Down
Loading