"""CLI interface for agenix."""
import os
import sys
from typing import Optional
from prompt_toolkit import PromptSession
from prompt_toolkit.styles import Style
from rich import box
from rich.align import Align
from rich.console import Console
from rich.layout import Layout
from rich.live import Live
from rich.markdown import Markdown
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from ..core.messages import (Event, MessageEndEvent, MessageStartEvent,
MessageUpdateEvent, TextContent,
ToolExecutionEndEvent, ToolExecutionStartEvent,
ToolExecutionUpdateEvent, TurnEndEvent,
TurnStartEvent)
[docs]
class CLIRenderer:
"""Render agent events to terminal."""
[docs]
def __init__(self):
self.console = Console()
self.current_message = ""
self.current_tool = None
self.message_buffer = [] # For buffering streaming content
self.in_live_mode = False
# Initialize prompt session with better unicode support
self.prompt_session = PromptSession(
style=Style.from_dict({
'prompt': '#0066cc bold', # Blue bold for "You:"
})
)
[docs]
def render_event(self, event: Event) -> None:
"""Render an event to the console."""
if isinstance(event, TurnStartEvent):
self.console.print(
"\n[dim]───────────────────────────────────[/dim]")
elif isinstance(event, MessageStartEvent):
self.current_message = ""
self.message_buffer = []
elif isinstance(event, MessageUpdateEvent):
# Accumulate message
self.current_message += event.delta
self.message_buffer.append(event.delta)
# Simple streaming output
self.console.print(event.delta, end="", markup=False)
elif isinstance(event, MessageEndEvent):
# Finalize message display
if self.current_message:
self.console.print() # New line
self.current_message = ""
self.message_buffer = []
elif isinstance(event, ToolExecutionStartEvent):
self.current_tool = event.tool_name
# Prettier tool execution start
tool_icon = {
"read": "📖",
"write": "✍️",
"edit": "✏️",
"bash": "⚡",
"grep": "🔍",
}.get(event.tool_name, "🔧")
self.console.print(
f"\n{tool_icon} [bold cyan]Executing:[/bold cyan] [yellow]{event.tool_name}[/yellow]"
)
if event.args:
# Show compact args
args_display = []
for k, v in list(event.args.items())[:2]:
v_str = str(v)
if len(v_str) > 50:
v_str = v_str[:47] + "..."
args_display.append(f"{k}={v_str}")
if len(event.args) > 2:
args_display.append("...")
self.console.print(f" [dim]{', '.join(args_display)}[/dim]")
elif isinstance(event, ToolExecutionUpdateEvent):
# Show tool progress (if any)
if event.partial_result:
update_text = str(event.partial_result)
if len(update_text) > 100:
update_text = update_text[:97] + "..."
self.console.print(f" [dim]{update_text}[/dim]")
elif isinstance(event, ToolExecutionEndEvent):
# Show tool result with status
if event.is_error:
self.console.print(f" [bold red]✗ Error[/bold red]")
error_text = str(event.result)
if len(error_text) > 300:
error_text = error_text[:297] + "..."
self.console.print(Panel(
error_text,
border_style="red",
title="[red]Tool Error[/red]",
box=box.ROUNDED
))
else:
self.console.print(f" [bold green]✓ Done[/bold green]")
self.current_tool = None
elif isinstance(event, TurnEndEvent):
# Show token usage
if event.message and event.message.usage:
usage = event.message.usage
total_tokens = usage.input_tokens + usage.output_tokens
self.console.print(
f"\n[dim]Tokens: {usage.input_tokens:,} in · "
f"{usage.output_tokens:,} out · "
f"{total_tokens:,} total[/dim]"
)
[docs]
def render_message(self, role: str, content: str, is_error: bool = False) -> None:
"""Render a complete message."""
if role == "user":
self.console.print(f"\n[bold blue]You:[/bold blue] {content}")
elif role == "assistant":
self.console.print(f"\n[bold green]Assistant:[/bold green]")
self.console.print(content)
elif role == "system":
style = "red" if is_error else "yellow"
self.console.print(
f"\n[bold {style}]System:[/bold {style}] {content}")
[docs]
def render_error(self, error: str) -> None:
"""Render an error."""
self.console.print(Panel(
f"[bold red]Error:[/bold red] {error}",
border_style="red"
))
[docs]
def render_welcome(self, tools=None) -> None:
"""Render welcome message with banner and system info."""
from .. import __version__
# ASCII Art Banner
banner = """
___ _
/ _ \\ (_)
/ /_\\ \\_ _ ___ _ __ ___ __
| _ | | | |/ _ \\ '_ \\| \\ \\/ /
| | | | |_| | __/ | | | |> <
\\_| |_/\\__, |\\___|_| |_|_/_/\\_\\
__/ |
|___/
"""
# Header with banner
header = Panel(
Align.center(
f"[bold cyan]{banner}[/bold cyan]\n"
f"[yellow]A lightweight AI coding agent[/yellow]\n"
f"[dim]Version {__version__} | MIT License[/dim]"
),
box=box.DOUBLE,
border_style="cyan",
padding=(1, 2)
)
self.console.print(header)
# System Information
info_table = Table(show_header=False, box=None, padding=(0, 2))
info_table.add_column(style="cyan bold")
info_table.add_column(style="white")
# Get API provider info
api_key = os.getenv("OPENAI_API_KEY") or os.getenv("ANTHROPIC_API_KEY")
provider = "OpenAI" if os.getenv("OPENAI_API_KEY") else "Anthropic" if os.getenv(
"ANTHROPIC_API_KEY") else "Not Set"
info_table.add_row("📍 Working Dir", os.getcwd())
info_table.add_row("🤖 API Provider", provider)
info_table.add_row(
"💾 Session Dir", os.path.expanduser("~/.agenix/sessions"))
self.console.print(Panel(
info_table,
title="[bold]System Information[/bold]",
border_style="blue",
box=box.ROUNDED
))
# Available Tools
if tools:
tools_table = Table(
show_header=True,
header_style="bold magenta",
box=box.SIMPLE_HEAD,
padding=(0, 1)
)
tools_table.add_column("🛠️ Tool", style="cyan", no_wrap=True)
tools_table.add_column("Description", style="white")
tool_descriptions = {
"read": "Read files and images",
"write": "Create or overwrite files",
"edit": "Surgical file edits",
"bash": "Execute shell commands",
"grep": "Search code patterns",
}
for tool in tools:
desc = tool_descriptions.get(tool.name, tool.description[:40])
tools_table.add_row(tool.name, desc)
self.console.print(Panel(
tools_table,
title="[bold]Available Tools[/bold]",
border_style="green",
box=box.ROUNDED
))
# Commands
commands_table = Table(
show_header=False,
box=None,
padding=(0, 1),
show_lines=False
)
commands_table.add_column(style="yellow bold", no_wrap=True)
commands_table.add_column(style="dim")
commands_table.add_row("/help", "Show this help message")
commands_table.add_row("/clear", "Clear conversation history")
commands_table.add_row("/sessions", "List saved sessions")
commands_table.add_row("/load <id>", "Load a previous session")
commands_table.add_row("/quit", "Exit the program")
self.console.print(Panel(
commands_table,
title="[bold]Commands[/bold]",
border_style="yellow",
box=box.ROUNDED
))
# Quick Start
self.console.print("\n[bold cyan]💡 Quick Start:[/bold cyan]")
self.console.print(
" • Type your message to start chatting with the AI agent")
self.console.print(
" • The agent can read/write files, execute commands, and more")
self.console.print(
" • Press [bold]Ctrl+C[/bold] anytime to interrupt\n")
[docs]
def prompt(self, text: str = "You") -> str:
"""Show input prompt with better unicode support."""
try:
# Use prompt_toolkit for better unicode handling (especially Chinese)
return self.prompt_session.prompt(f"\n{text}: ")
except EOFError:
return "/quit"
except KeyboardInterrupt:
return "/quit"
[docs]
def clear(self) -> None:
"""Clear the screen."""
self.console.clear()
[docs]
class CLI:
"""Main CLI interface."""
[docs]
def __init__(self, renderer: Optional[CLIRenderer] = None):
self.renderer = renderer or CLIRenderer()
self.tools = None # Store tools for /help command
[docs]
def run_interactive(self, agent, tools=None) -> None:
"""Run interactive chat loop."""
import asyncio
self.tools = tools # Store for /help command
self.renderer.render_welcome(tools=tools)
while True:
try:
# Get user input
user_input = self.renderer.prompt()
if not user_input.strip():
continue
# Handle commands
if user_input.startswith("/"):
if not self.handle_command(user_input, agent):
break
continue
# Process message
asyncio.run(self.process_message(agent, user_input))
except KeyboardInterrupt:
self.renderer.render_message("system", "\nUse /quit to exit")
continue
except Exception as e:
self.renderer.render_error(str(e))
[docs]
async def process_message(self, agent, user_input: str) -> None:
"""Process a user message."""
try:
# Stream agent response
async for event in agent.prompt(user_input):
self.renderer.render_event(event)
except KeyboardInterrupt:
self.renderer.render_message("system", "\nInterrupted by user")
except Exception as e:
import traceback
error_msg = f"Error processing message: {str(e)}"
if "--debug" in sys.argv:
error_msg += f"\n\nTraceback:\n{traceback.format_exc()}"
self.renderer.render_error(error_msg)
[docs]
def handle_command(self, command: str, agent) -> bool:
"""Handle CLI commands.
Returns:
True to continue, False to exit
"""
parts = command.split(maxsplit=1)
cmd = parts[0].lower()
args = parts[1] if len(parts) > 1 else ""
if cmd in ["/quit", "/exit"]:
self.renderer.render_message("system", "Goodbye! 👋")
return False
elif cmd == "/help":
self.renderer.render_welcome(tools=self.tools)
elif cmd == "/clear":
agent.clear_messages()
self.renderer.clear()
self.renderer.render_message("system", "Conversation cleared")
elif cmd == "/sessions":
self.list_sessions()
elif cmd == "/load":
if args:
self.load_session(agent, args)
else:
self.renderer.render_message(
"system", "Usage: /load <session_id>", is_error=True)
else:
self.renderer.render_message(
"system", f"Unknown command: {cmd}", is_error=True)
return True
[docs]
def list_sessions(self) -> None:
"""List saved sessions."""
from ..core.session import SessionManager
manager = SessionManager()
sessions = manager.list_sessions()
if not sessions:
self.renderer.render_message("system", "No saved sessions")
return
self.renderer.console.print("\n[bold]Saved Sessions:[/bold]")
for session in sessions:
self.renderer.console.print(
f" • {session['session_id']} - {session['created_at']}"
)
[docs]
def load_session(self, agent, session_id: str) -> None:
"""Load a session."""
from ..core.session import SessionManager
try:
manager = SessionManager()
messages = manager.load_session(session_id)
agent.messages = messages
self.renderer.render_message(
"system",
f"Loaded session: {session_id} ({len(messages)} messages)"
)
except FileNotFoundError:
self.renderer.render_error(f"Session not found: {session_id}")
except Exception as e:
self.renderer.render_error(f"Error loading session: {str(e)}")