ilia/nanobot
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Trim CLI patch to remove unrelated whitespace churn

Browse Source
This commit is contained in:
Chris Alexander 2026-02-08 21:07:02 +00:00
parent 0a2d557268
commit 9c6ffa0d56
No known key found for this signature in database
1 changed files with 126 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

128
nanobot/cli/commands.py
View File

@ -26,29 +26,6 @@ app = typer.Typer(
console = Console() console = Console()
EXIT_COMMANDS = {"exit", "quit", "/exit", "/quit", ":q"} 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 # Lightweight CLI input: readline for arrow keys / history, termios for flush
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ -71,7 +48,6 @@ def _flush_pending_tty_input() -> None:
try: try:
import termios import termios
termios.tcflush(fd, termios.TCIFLUSH) termios.tcflush(fd, termios.TCIFLUSH)
return return
except Exception: except Exception:
@ -103,7 +79,6 @@ def _restore_terminal() -> None:
return return
try: try:
import termios import termios
termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, _SAVED_TERM_ATTRS) termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, _SAVED_TERM_ATTRS)
except Exception: except Exception:
pass pass
@ -116,7 +91,6 @@ def _enable_line_editing() -> None:
# Save terminal state before readline touches it # Save terminal state before readline touches it
try: try:
import termios import termios
_SAVED_TERM_ATTRS = termios.tcgetattr(sys.stdin.fileno()) _SAVED_TERM_ATTRS = termios.tcgetattr(sys.stdin.fileno())
except Exception: except Exception:
pass pass
@ -162,6 +136,28 @@ def _prompt_text() -> str:
return "\001\033[1;34m\002You:\001\033[0m\002 " return "\001\033[1;34m\002You:\001\033[0m\002 "
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
async def _read_interactive_input_async() -> str: async def _read_interactive_input_async() -> str:
"""Read user input with arrow keys and history (runs input() in a thread).""" """Read user input with arrow keys and history (runs input() in a thread)."""
try: try:
@ -178,7 +174,9 @@ def version_callback(value: bool):
@app.callback() @app.callback()
def main( 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.""" """nanobot - Personal AI Assistant."""
pass pass
@ -219,10 +217,10 @@ def onboard():
console.print("\nNext steps:") console.print("\nNext steps:")
console.print(" 1. Add your API key to [cyan]~/.nanobot/config.json[/cyan]") console.print(" 1. Add your API key to [cyan]~/.nanobot/config.json[/cyan]")
console.print(" Get one at: https://openrouter.ai/keys") console.print(" Get one at: https://openrouter.ai/keys")
console.print(' 2. Chat: [cyan]nanobot agent -m "Hello!"[/cyan]') console.print(" 2. Chat: [cyan]nanobot agent -m \"Hello!\"[/cyan]")
console.print( console.print("\n[dim]Want Telegram/WhatsApp? See: https://github.com/HKUDS/nanobot#-chat-apps[/dim]")
"\n[dim]Want Telegram/WhatsApp? See: https://github.com/HKUDS/nanobot#-chat-apps[/dim]"
)
def _create_workspace_templates(workspace: Path): def _create_workspace_templates(workspace: Path):
@ -300,7 +298,6 @@ This file stores important information that should persist across sessions.
def _make_provider(config): def _make_provider(config):
"""Create LiteLLMProvider from config. Exits if no API key found.""" """Create LiteLLMProvider from config. Exits if no API key found."""
from nanobot.providers.litellm_provider import LiteLLMProvider from nanobot.providers.litellm_provider import LiteLLMProvider
p = config.get_provider() p = config.get_provider()
model = config.agents.defaults.model model = config.agents.defaults.model
if not (p and p.api_key) and not model.startswith("bedrock/"): if not (p and p.api_key) and not model.startswith("bedrock/"):
@ -338,7 +335,6 @@ def gateway(
if verbose: if verbose:
import logging import logging
logging.basicConfig(level=logging.DEBUG) logging.basicConfig(level=logging.DEBUG)
console.print(f"{__logo__} Starting nanobot gateway on port {port}...") console.print(f"{__logo__} Starting nanobot gateway on port {port}...")
@ -377,16 +373,12 @@ def gateway(
) )
if job.payload.deliver and job.payload.to: if job.payload.deliver and job.payload.to:
from nanobot.bus.events import OutboundMessage from nanobot.bus.events import OutboundMessage
await bus.publish_outbound(OutboundMessage(
await bus.publish_outbound( channel=job.payload.channel or "cli",
OutboundMessage( chat_id=job.payload.to,
channel=job.payload.channel or "cli", content=response or ""
chat_id=job.payload.to, ))
content=response or "",
)
)
return response return response
cron.on_job = on_cron_job cron.on_job = on_cron_job
# Create heartbeat service # Create heartbeat service
@ -398,7 +390,7 @@ def gateway(
workspace=config.workspace_path, workspace=config.workspace_path,
on_heartbeat=on_heartbeat, on_heartbeat=on_heartbeat,
interval_s=30 * 60, # 30 minutes interval_s=30 * 60, # 30 minutes
enabled=True, enabled=True
) )
# Create channel manager # Create channel manager
@ -433,6 +425,8 @@ def gateway(
asyncio.run(run()) asyncio.run(run())
# ============================================================================ # ============================================================================
# Agent Commands # Agent Commands
# ============================================================================ # ============================================================================
@ -442,12 +436,8 @@ def gateway(
def agent( def agent(
message: str = typer.Option(None, "--message", "-m", help="Message to send to the 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"), session_id: str = typer.Option("cli:default", "--session", "-s", help="Session ID"),
markdown: bool = typer.Option( markdown: bool = typer.Option(True, "--markdown/--no-markdown", help="Render assistant output as Markdown"),
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"),
),
logs: bool = typer.Option(
False, "--logs/--no-logs", help="Show nanobot runtime logs during chat"
),
): ):
"""Interact with the agent directly.""" """Interact with the agent directly."""
from nanobot.config.loader import load_config from nanobot.config.loader import load_config
@ -548,15 +538,27 @@ def channels_status():
# WhatsApp # WhatsApp
wa = config.channels.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 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 # Telegram
tg = config.channels.telegram tg = config.channels.telegram
tg_config = f"token: {tg.token[:10]}..." if tg.token else "[dim]not configured[/dim]" 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) console.print(table)
@ -670,7 +672,6 @@ def cron_list(
table.add_column("Next Run") table.add_column("Next Run")
import time import time
for job in jobs: for job in jobs:
# Format schedule # Format schedule
if job.schedule.kind == "every": if job.schedule.kind == "every":
@ -683,9 +684,7 @@ def cron_list(
# Format next run # Format next run
next_run = "" next_run = ""
if job.state.next_run_at_ms: if job.state.next_run_at_ms:
next_time = time.strftime( next_time = time.strftime("%Y-%m-%d %H:%M", time.localtime(job.state.next_run_at_ms / 1000))
"%Y-%m-%d %H:%M", time.localtime(job.state.next_run_at_ms / 1000)
)
next_run = next_time next_run = next_time
status = "[green]enabled[/green]" if job.enabled else "[dim]disabled[/dim]" status = "[green]enabled[/green]" if job.enabled else "[dim]disabled[/dim]"
@ -704,9 +703,7 @@ def cron_add(
at: str = typer.Option(None, "--at", help="Run once at time (ISO format)"), 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"), deliver: bool = typer.Option(False, "--deliver", "-d", help="Deliver response to channel"),
to: str = typer.Option(None, "--to", help="Recipient for delivery"), to: str = typer.Option(None, "--to", help="Recipient for delivery"),
channel: str = typer.Option( channel: str = typer.Option(None, "--channel", help="Channel for delivery (e.g. 'telegram', 'whatsapp')"),
None, "--channel", help="Channel for delivery (e.g. 'telegram', 'whatsapp')"
),
): ):
"""Add a scheduled job.""" """Add a scheduled job."""
from nanobot.config.loader import get_data_dir from nanobot.config.loader import get_data_dir
@ -720,7 +717,6 @@ def cron_add(
schedule = CronSchedule(kind="cron", expr=cron_expr) schedule = CronSchedule(kind="cron", expr=cron_expr)
elif at: elif at:
import datetime import datetime
dt = datetime.datetime.fromisoformat(at) dt = datetime.datetime.fromisoformat(at)
schedule = CronSchedule(kind="at", at_ms=int(dt.timestamp() * 1000)) schedule = CronSchedule(kind="at", at_ms=int(dt.timestamp() * 1000))
else: else:
@ -816,12 +812,8 @@ def status():
console.print(f"{__logo__} nanobot Status\n") console.print(f"{__logo__} nanobot Status\n")
console.print( console.print(f"Config: {config_path} {'[green]✓[/green]' if config_path.exists() else '[red]✗[/red]'}")
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"Workspace: {workspace} {'[green]✓[/green]' if workspace.exists() else '[red]✗[/red]'}"
)
if config_path.exists(): if config_path.exists():
from nanobot.providers.registry import PROVIDERS from nanobot.providers.registry import PROVIDERS
@ -841,9 +833,7 @@ def status():
console.print(f"{spec.label}: [dim]not set[/dim]") console.print(f"{spec.label}: [dim]not set[/dim]")
else: else:
has_key = bool(p.api_key) has_key = bool(p.api_key)
console.print( console.print(f"{spec.label}: {'[green]✓[/green]' if has_key else '[dim]not set[/dim]'}")
f"{spec.label}: {'[green]✓[/green]' if has_key else '[dim]not set[/dim]'}"
)
if __name__ == "__main__": if __name__ == "__main__":