Improve agent CLI chat UX with markdown output and clearer interaction feedback

This commit is contained in:
Chris Alexander 2026-02-08 20:50:31 +00:00
parent 8af98004b3
commit 0a2d557268
No known key found for this signature in database

View File

@ -10,7 +10,10 @@ import sys
import typer
from rich.console import Console
from rich.markdown import Markdown
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from nanobot import __version__, __logo__
@ -21,6 +24,30 @@ app = typer.Typer(
)
console = Console()
EXIT_COMMANDS = {"exit", "quit", "/exit", "/quit", ":q"}
def _print_agent_response(response: str, render_markdown: bool) -> None:
"""Render assistant response with consistent terminal styling."""
content = response or ""
body = Markdown(content) if render_markdown else Text(content)
console.print()
console.print(
Panel(
body,
title=f"{__logo__} Nanobot",
title_align="left",
border_style="cyan",
padding=(0, 1),
)
)
console.print()
def _is_exit_command(command: str) -> bool:
"""Return True when input should end interactive chat."""
return command.lower() in EXIT_COMMANDS
# ---------------------------------------------------------------------------
# Lightweight CLI input: readline for arrow keys / history, termios for flush
@ -44,6 +71,7 @@ def _flush_pending_tty_input() -> None:
try:
import termios
termios.tcflush(fd, termios.TCIFLUSH)
return
except Exception:
@ -75,6 +103,7 @@ def _restore_terminal() -> None:
return
try:
import termios
termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, _SAVED_TERM_ATTRS)
except Exception:
pass
@ -87,6 +116,7 @@ def _enable_line_editing() -> None:
# Save terminal state before readline touches it
try:
import termios
_SAVED_TERM_ATTRS = termios.tcgetattr(sys.stdin.fileno())
except Exception:
pass
@ -148,9 +178,7 @@ def version_callback(value: bool):
@app.callback()
def main(
version: bool = typer.Option(
None, "--version", "-v", callback=version_callback, is_eager=True
),
version: bool = typer.Option(None, "--version", "-v", callback=version_callback, is_eager=True),
):
"""nanobot - Personal AI Assistant."""
pass
@ -167,34 +195,34 @@ def onboard():
from nanobot.config.loader import get_config_path, save_config
from nanobot.config.schema import Config
from nanobot.utils.helpers import get_workspace_path
config_path = get_config_path()
if config_path.exists():
console.print(f"[yellow]Config already exists at {config_path}[/yellow]")
if not typer.confirm("Overwrite?"):
raise typer.Exit()
# Create default config
config = Config()
save_config(config)
console.print(f"[green]✓[/green] Created config at {config_path}")
# Create workspace
workspace = get_workspace_path()
console.print(f"[green]✓[/green] Created workspace at {workspace}")
# Create default bootstrap files
_create_workspace_templates(workspace)
console.print(f"\n{__logo__} nanobot is ready!")
console.print("\nNext steps:")
console.print(" 1. Add your API key to [cyan]~/.nanobot/config.json[/cyan]")
console.print(" Get one at: https://openrouter.ai/keys")
console.print(" 2. Chat: [cyan]nanobot agent -m \"Hello!\"[/cyan]")
console.print("\n[dim]Want Telegram/WhatsApp? See: https://github.com/HKUDS/nanobot#-chat-apps[/dim]")
console.print(' 2. Chat: [cyan]nanobot agent -m "Hello!"[/cyan]')
console.print(
"\n[dim]Want Telegram/WhatsApp? See: https://github.com/HKUDS/nanobot#-chat-apps[/dim]"
)
def _create_workspace_templates(workspace: Path):
@ -238,13 +266,13 @@ Information about the user goes here.
- Language: (your preferred language)
""",
}
for filename, content in templates.items():
file_path = workspace / filename
if not file_path.exists():
file_path.write_text(content)
console.print(f" [dim]Created {filename}[/dim]")
# Create memory directory and MEMORY.md
memory_dir = workspace / "memory"
memory_dir.mkdir(exist_ok=True)
@ -272,6 +300,7 @@ This file stores important information that should persist across sessions.
def _make_provider(config):
"""Create LiteLLMProvider from config. Exits if no API key found."""
from nanobot.providers.litellm_provider import LiteLLMProvider
p = config.get_provider()
model = config.agents.defaults.model
if not (p and p.api_key) and not model.startswith("bedrock/"):
@ -306,22 +335,23 @@ def gateway(
from nanobot.cron.service import CronService
from nanobot.cron.types import CronJob
from nanobot.heartbeat.service import HeartbeatService
if verbose:
import logging
logging.basicConfig(level=logging.DEBUG)
console.print(f"{__logo__} Starting nanobot gateway on port {port}...")
config = load_config()
bus = MessageBus()
provider = _make_provider(config)
session_manager = SessionManager(config.workspace_path)
# Create cron service first (callback set after agent creation)
cron_store_path = get_data_dir() / "cron" / "jobs.json"
cron = CronService(cron_store_path)
# Create agent with cron service
agent = AgentLoop(
bus=bus,
@ -335,7 +365,7 @@ def gateway(
restrict_to_workspace=config.tools.restrict_to_workspace,
session_manager=session_manager,
)
# Set cron callback (needs agent)
async def on_cron_job(job: CronJob) -> str | None:
"""Execute a cron job through the agent."""
@ -347,40 +377,44 @@ def gateway(
)
if job.payload.deliver and job.payload.to:
from nanobot.bus.events import OutboundMessage
await bus.publish_outbound(OutboundMessage(
channel=job.payload.channel or "cli",
chat_id=job.payload.to,
content=response or ""
))
await bus.publish_outbound(
OutboundMessage(
channel=job.payload.channel or "cli",
chat_id=job.payload.to,
content=response or "",
)
)
return response
cron.on_job = on_cron_job
# Create heartbeat service
async def on_heartbeat(prompt: str) -> str:
"""Execute heartbeat through the agent."""
return await agent.process_direct(prompt, session_key="heartbeat")
heartbeat = HeartbeatService(
workspace=config.workspace_path,
on_heartbeat=on_heartbeat,
interval_s=30 * 60, # 30 minutes
enabled=True
enabled=True,
)
# Create channel manager
channels = ChannelManager(config, bus, session_manager=session_manager)
if channels.enabled_channels:
console.print(f"[green]✓[/green] Channels enabled: {', '.join(channels.enabled_channels)}")
else:
console.print("[yellow]Warning: No channels enabled[/yellow]")
cron_status = cron.status()
if cron_status["jobs"] > 0:
console.print(f"[green]✓[/green] Cron: {cron_status['jobs']} scheduled jobs")
console.print(f"[green]✓[/green] Heartbeat: every 30m")
async def run():
try:
await cron.start()
@ -395,12 +429,10 @@ def gateway(
cron.stop()
agent.stop()
await channels.stop_all()
asyncio.run(run())
# ============================================================================
# Agent Commands
# ============================================================================
@ -410,17 +442,29 @@ def gateway(
def agent(
message: str = typer.Option(None, "--message", "-m", help="Message to send to the agent"),
session_id: str = typer.Option("cli:default", "--session", "-s", help="Session ID"),
markdown: bool = typer.Option(
True, "--markdown/--no-markdown", help="Render assistant output as Markdown"
),
logs: bool = typer.Option(
False, "--logs/--no-logs", help="Show nanobot runtime logs during chat"
),
):
"""Interact with the agent directly."""
from nanobot.config.loader import load_config
from nanobot.bus.queue import MessageBus
from nanobot.agent.loop import AgentLoop
from loguru import logger
config = load_config()
bus = MessageBus()
provider = _make_provider(config)
if logs:
logger.enable("nanobot")
else:
logger.disable("nanobot")
agent_loop = AgentLoop(
bus=bus,
provider=provider,
@ -429,13 +473,14 @@ def agent(
exec_config=config.tools.exec,
restrict_to_workspace=config.tools.restrict_to_workspace,
)
if message:
# Single message mode
async def run_once():
response = await agent_loop.process_direct(message, session_id)
console.print(f"\n{__logo__} {response}")
with console.status("[dim]Nanobot is thinking...[/dim]", spinner="dots"):
response = await agent_loop.process_direct(message, session_id)
_print_agent_response(response, render_markdown=markdown)
asyncio.run(run_once())
else:
# Interactive mode
@ -451,23 +496,32 @@ def agent(
os._exit(0)
signal.signal(signal.SIGINT, _exit_on_sigint)
async def run_interactive():
while True:
try:
_flush_pending_tty_input()
user_input = await _read_interactive_input_async()
if not user_input.strip():
command = user_input.strip()
if not command:
continue
response = await agent_loop.process_direct(user_input, session_id)
console.print(f"\n{__logo__} {response}\n")
if _is_exit_command(command):
console.print("\nGoodbye!")
break
with console.status("[dim]Nanobot is thinking...[/dim]", spinner="dots"):
response = await agent_loop.process_direct(user_input, session_id)
_print_agent_response(response, render_markdown=markdown)
except KeyboardInterrupt:
_save_history()
_restore_terminal()
console.print("\nGoodbye!")
break
except EOFError:
console.print("\nGoodbye!")
break
asyncio.run(run_interactive())
@ -494,27 +548,15 @@ def channels_status():
# WhatsApp
wa = config.channels.whatsapp
table.add_row(
"WhatsApp",
"" if wa.enabled else "",
wa.bridge_url
)
table.add_row("WhatsApp", "" if wa.enabled else "", wa.bridge_url)
dc = config.channels.discord
table.add_row(
"Discord",
"" if dc.enabled else "",
dc.gateway_url
)
table.add_row("Discord", "" if dc.enabled else "", dc.gateway_url)
# Telegram
tg = config.channels.telegram
tg_config = f"token: {tg.token[:10]}..." if tg.token else "[dim]not configured[/dim]"
table.add_row(
"Telegram",
"" if tg.enabled else "",
tg_config
)
table.add_row("Telegram", "" if tg.enabled else "", tg_config)
console.print(table)
@ -523,57 +565,57 @@ def _get_bridge_dir() -> Path:
"""Get the bridge directory, setting it up if needed."""
import shutil
import subprocess
# User's bridge location
user_bridge = Path.home() / ".nanobot" / "bridge"
# Check if already built
if (user_bridge / "dist" / "index.js").exists():
return user_bridge
# Check for npm
if not shutil.which("npm"):
console.print("[red]npm not found. Please install Node.js >= 18.[/red]")
raise typer.Exit(1)
# Find source bridge: first check package data, then source dir
pkg_bridge = Path(__file__).parent.parent / "bridge" # nanobot/bridge (installed)
src_bridge = Path(__file__).parent.parent.parent / "bridge" # repo root/bridge (dev)
source = None
if (pkg_bridge / "package.json").exists():
source = pkg_bridge
elif (src_bridge / "package.json").exists():
source = src_bridge
if not source:
console.print("[red]Bridge source not found.[/red]")
console.print("Try reinstalling: pip install --force-reinstall nanobot")
raise typer.Exit(1)
console.print(f"{__logo__} Setting up bridge...")
# Copy to user directory
user_bridge.parent.mkdir(parents=True, exist_ok=True)
if user_bridge.exists():
shutil.rmtree(user_bridge)
shutil.copytree(source, user_bridge, ignore=shutil.ignore_patterns("node_modules", "dist"))
# Install and build
try:
console.print(" Installing dependencies...")
subprocess.run(["npm", "install"], cwd=user_bridge, check=True, capture_output=True)
console.print(" Building...")
subprocess.run(["npm", "run", "build"], cwd=user_bridge, check=True, capture_output=True)
console.print("[green]✓[/green] Bridge ready\n")
except subprocess.CalledProcessError as e:
console.print(f"[red]Build failed: {e}[/red]")
if e.stderr:
console.print(f"[dim]{e.stderr.decode()[:500]}[/dim]")
raise typer.Exit(1)
return user_bridge
@ -581,12 +623,12 @@ def _get_bridge_dir() -> Path:
def channels_login():
"""Link device via QR code."""
import subprocess
bridge_dir = _get_bridge_dir()
console.print(f"{__logo__} Starting bridge...")
console.print("Scan the QR code to connect.\n")
try:
subprocess.run(["npm", "start"], cwd=bridge_dir, check=True)
except subprocess.CalledProcessError as e:
@ -610,24 +652,25 @@ def cron_list(
"""List scheduled jobs."""
from nanobot.config.loader import get_data_dir
from nanobot.cron.service import CronService
store_path = get_data_dir() / "cron" / "jobs.json"
service = CronService(store_path)
jobs = service.list_jobs(include_disabled=all)
if not jobs:
console.print("No scheduled jobs.")
return
table = Table(title="Scheduled Jobs")
table.add_column("ID", style="cyan")
table.add_column("Name")
table.add_column("Schedule")
table.add_column("Status")
table.add_column("Next Run")
import time
for job in jobs:
# Format schedule
if job.schedule.kind == "every":
@ -636,17 +679,19 @@ def cron_list(
sched = job.schedule.expr or ""
else:
sched = "one-time"
# Format next run
next_run = ""
if job.state.next_run_at_ms:
next_time = time.strftime("%Y-%m-%d %H:%M", time.localtime(job.state.next_run_at_ms / 1000))
next_time = time.strftime(
"%Y-%m-%d %H:%M", time.localtime(job.state.next_run_at_ms / 1000)
)
next_run = next_time
status = "[green]enabled[/green]" if job.enabled else "[dim]disabled[/dim]"
table.add_row(job.id, job.name, sched, status, next_run)
console.print(table)
@ -659,13 +704,15 @@ def cron_add(
at: str = typer.Option(None, "--at", help="Run once at time (ISO format)"),
deliver: bool = typer.Option(False, "--deliver", "-d", help="Deliver response to channel"),
to: str = typer.Option(None, "--to", help="Recipient for delivery"),
channel: str = typer.Option(None, "--channel", help="Channel for delivery (e.g. 'telegram', 'whatsapp')"),
channel: str = typer.Option(
None, "--channel", help="Channel for delivery (e.g. 'telegram', 'whatsapp')"
),
):
"""Add a scheduled job."""
from nanobot.config.loader import get_data_dir
from nanobot.cron.service import CronService
from nanobot.cron.types import CronSchedule
# Determine schedule type
if every:
schedule = CronSchedule(kind="every", every_ms=every * 1000)
@ -673,15 +720,16 @@ def cron_add(
schedule = CronSchedule(kind="cron", expr=cron_expr)
elif at:
import datetime
dt = datetime.datetime.fromisoformat(at)
schedule = CronSchedule(kind="at", at_ms=int(dt.timestamp() * 1000))
else:
console.print("[red]Error: Must specify --every, --cron, or --at[/red]")
raise typer.Exit(1)
store_path = get_data_dir() / "cron" / "jobs.json"
service = CronService(store_path)
job = service.add_job(
name=name,
schedule=schedule,
@ -690,7 +738,7 @@ def cron_add(
to=to,
channel=channel,
)
console.print(f"[green]✓[/green] Added job '{job.name}' ({job.id})")
@ -701,10 +749,10 @@ def cron_remove(
"""Remove a scheduled job."""
from nanobot.config.loader import get_data_dir
from nanobot.cron.service import CronService
store_path = get_data_dir() / "cron" / "jobs.json"
service = CronService(store_path)
if service.remove_job(job_id):
console.print(f"[green]✓[/green] Removed job {job_id}")
else:
@ -719,10 +767,10 @@ def cron_enable(
"""Enable or disable a job."""
from nanobot.config.loader import get_data_dir
from nanobot.cron.service import CronService
store_path = get_data_dir() / "cron" / "jobs.json"
service = CronService(store_path)
job = service.enable_job(job_id, enabled=not disable)
if job:
status = "disabled" if disable else "enabled"
@ -739,13 +787,13 @@ def cron_run(
"""Manually run a job."""
from nanobot.config.loader import get_data_dir
from nanobot.cron.service import CronService
store_path = get_data_dir() / "cron" / "jobs.json"
service = CronService(store_path)
async def run():
return await service.run_job(job_id, force=force)
if asyncio.run(run()):
console.print(f"[green]✓[/green] Job executed")
else:
@ -768,14 +816,18 @@ def status():
console.print(f"{__logo__} nanobot Status\n")
console.print(f"Config: {config_path} {'[green]✓[/green]' if config_path.exists() else '[red]✗[/red]'}")
console.print(f"Workspace: {workspace} {'[green]✓[/green]' if workspace.exists() else '[red]✗[/red]'}")
console.print(
f"Config: {config_path} {'[green]✓[/green]' if config_path.exists() else '[red]✗[/red]'}"
)
console.print(
f"Workspace: {workspace} {'[green]✓[/green]' if workspace.exists() else '[red]✗[/red]'}"
)
if config_path.exists():
from nanobot.providers.registry import PROVIDERS
console.print(f"Model: {config.agents.defaults.model}")
# Check API keys from registry
for spec in PROVIDERS:
p = getattr(config.providers, spec.name, None)
@ -789,7 +841,9 @@ def status():
console.print(f"{spec.label}: [dim]not set[/dim]")
else:
has_key = bool(p.api_key)
console.print(f"{spec.label}: {'[green]✓[/green]' if has_key else '[dim]not set[/dim]'}")
console.print(
f"{spec.label}: {'[green]✓[/green]' if has_key else '[dim]not set[/dim]'}"
)
if __name__ == "__main__":