#!/usr/bin/env python3 """ Simple CLI tool to converse with the Organon ecosystem group. """ import os import sys from dotenv import load_dotenv from letta_client import Letta from rich.console import Console from rich.prompt import Prompt from rich.panel import Panel from rich.text import Text # Add parent directory to path for imports sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config_loader import get_config load_dotenv() class OrganonChat: def __init__(self): """Initialize the Organon chat client.""" self.console = Console() self.config = get_config() # Initialize Letta client self.letta_client = Letta( base_url=self.config.get('letta.base_url', os.environ.get('LETTA_BASE_URL')), token=self.config.get('letta.api_key', os.environ.get('LETTA_API_KEY')), timeout=self.config.get('letta.timeout', 30) ) # Get project ID self.project_id = self.config.get('letta.project_id', os.environ.get('LETTA_PROJECT_ID')) if not self.project_id: raise ValueError("Project ID must be set in config.yaml under letta.project_id or as LETTA_PROJECT_ID environment variable") # Find the Organon ecosystem group self.group_id = self._find_organon_group() self.console.print(Panel.fit( "[bold green]Organon Ecosystem Chat[/bold green]\n" f"Connected to group: {self.group_id}\n" "Type 'exit' or 'quit' to leave, 'help' for commands", title="🧠 Organon Chat" )) def _find_organon_group(self) -> str: """Find the Organon ecosystem group.""" try: self.console.print("[dim]Searching for Organon ecosystem group...[/dim]") # First, get the organon-central agent ID organon_agents = self.letta_client.agents.list(name="organon-central") if not organon_agents: raise ValueError("Organon central agent not found. Run create_organon.py first.") organon_central_id = organon_agents[0].id self.console.print(f"[dim]Found organon-central: {organon_central_id[:8]}[/dim]") # Get all groups (with and without project_id filter) try: groups = self.letta_client.groups.list() self.console.print(f"[dim]Found {len(groups)} groups with project filter[/dim]") except: try: groups = self.letta_client.groups.list() self.console.print(f"[dim]Found {len(groups)} groups without project filter[/dim]") except: groups = [] self.console.print("[dim]No groups found[/dim]") # Look for groups with organon-central as supervisor for group in groups: if hasattr(group, 'manager_config') and group.manager_config: if hasattr(group.manager_config, 'manager_agent_id') and group.manager_config.manager_agent_id == organon_central_id: self.console.print(f"[dim]Found Organon group: {group.id[:12]}[/dim]") return group.id # Look for the organon-ecosystem group by description for group in groups: if hasattr(group, 'description') and group.description and 'organon ecosystem' in group.description.lower(): self.console.print(f"[dim]Found Organon group by description: {group.id[:12]}[/dim]") return group.id # If still not found, try to find any group with organon shards shard_agents = self.letta_client.agents.list(project_id=self.project_id, tags=["organon-shard"]) if shard_agents: shard_ids = {shard.id for shard in shard_agents} for group in groups: try: members = self.letta_client.groups.agents.list(group_id=group.id) member_ids = {member.id for member in members} if shard_ids & member_ids: # If any shard is in this group self.console.print(f"[dim]Found group with Organon shards: {group.id[:12]}[/dim]") return group.id except: continue raise ValueError("Organon ecosystem group not found. Run 'python organon/setup_group.py' to create the group.") except Exception as e: raise ValueError(f"Error finding Organon group: {e}") def _display_response(self, response): """Display the group response in a formatted way.""" if hasattr(response, 'messages') and response.messages: for i, message in enumerate(response.messages): # Determine the sender sender = "Unknown" if hasattr(message, 'agent_id'): # Try to get agent name try: agent = self.letta_client.agents.retrieve(agent_id=message.agent_id) sender = agent.name if hasattr(agent, 'name') else f"Agent {message.agent_id[:8]}" except: sender = f"Agent {message.agent_id[:8]}" # Get message content content = "" if hasattr(message, 'text'): content = message.text elif hasattr(message, 'content'): content = message.content elif hasattr(message, 'message'): content = message.message elif message.message_type == "tool_return_message" and message.name == "send_message_to_all_agents_in_group": content = "Group responses:" try: # Parse the string representation of the list import ast responses = ast.literal_eval(message.tool_return) # Add each response to the content for response in responses: content += f"\n - {response}" except (ValueError, SyntaxError): # Fallback if parsing fails content += f"\n - {message.tool_return}" else: content = str(message) # Color based on sender type if "central" in sender.lower(): color = "blue" icon = "🧠" elif "shard" in sender.lower(): color = "cyan" icon = "🔹" else: color = "white" icon = "💬" self.console.print(Panel( Text(content, style=color), title=f"{icon} {sender}", border_style=color )) else: self.console.print("[yellow]No response messages received[/yellow]") def run(self): """Run the interactive chat loop.""" try: while True: # Get user input user_input = Prompt.ask("\n[bold green]You[/bold green]") # Handle special commands if user_input.lower() in ['exit', 'quit', 'q']: self.console.print("[yellow]Goodbye![/yellow]") break elif user_input.lower() == 'help': self.console.print(Panel( "[bold]Available commands:[/bold]\n" "• Type any message to send to the Organon ecosystem\n" "• 'help' - Show this help message\n" "• 'info' - Show group information\n" "• 'exit', 'quit', 'q' - Exit the chat", title="Help" )) continue elif user_input.lower() == 'info': self._show_group_info() continue elif not user_input.strip(): continue # Send message to group self.console.print("[dim]Sending to Organon ecosystem...[/dim]") try: response = self.letta_client.groups.messages.create( group_id=self.group_id, messages=[{ "role": "user", "content": user_input }] ) self.console.print("\n[bold]Organon Ecosystem Response:[/bold]") self._display_response(response) except Exception as e: self.console.print(f"[red]Error sending message: {e}[/red]") except KeyboardInterrupt: self.console.print("\n[yellow]Chat interrupted. Goodbye![/yellow]") except Exception as e: self.console.print(f"[red]Unexpected error: {e}[/red]") def _show_group_info(self): """Show information about the Organon group.""" try: group = self.letta_client.groups.retrieve(group_id=self.group_id) agents = self.letta_client.groups.agents.list(group_id=self.group_id) info_text = f"[bold]Group ID:[/bold] {self.group_id}\n" if hasattr(group, 'description'): info_text += f"[bold]Description:[/bold] {group.description}\n" info_text += f"[bold]Worker Agents:[/bold] {len(agents)}\n" for agent in agents: try: agent_detail = self.letta_client.agents.retrieve(agent_id=agent.id) name = agent_detail.name if hasattr(agent_detail, 'name') else agent.id[:8] info_text += f" • {name}\n" except: info_text += f" • {agent.id[:8]}\n" # Show supervisor info if hasattr(group, 'manager_config') and group.manager_config: if hasattr(group.manager_config, 'manager_agent_id'): try: supervisor = self.letta_client.agents.retrieve(agent_id=group.manager_config.manager_agent_id) supervisor_name = supervisor.name if hasattr(supervisor, 'name') else group.manager_config.manager_agent_id[:8] info_text += f"[bold]Supervisor:[/bold] {supervisor_name}\n" except: info_text += f"[bold]Supervisor:[/bold] {group.manager_config.manager_agent_id[:8]}\n" self.console.print(Panel(info_text, title="Group Information")) except Exception as e: self.console.print(f"[red]Error getting group info: {e}[/red]") def main(): """Main function.""" try: chat = OrganonChat() chat.run() except Exception as e: console = Console() console.print(f"[red]Failed to initialize chat: {e}[/red]") sys.exit(1) if __name__ == "__main__": main()