jpmschweitzer/webber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

feat: add config file support to CLI

Browse Source 
- Add ~/.webber/config.toml for persistent settings
- Support api.url, api.key, cli.mode, cli.stream, history.file
- Environment variables override config file values
- Add 'config' command to show settings and init config file
- Update all commands to use config defaults

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2026-01-14 12:06:13 +01:00
co-authored by Claude Opus 4.5
parent daa9543790
commit b87e61248e
4 changed files with 260 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
webber-cli/webber_cli/main.py
+97 -31
View File
@@ -8,7 +8,6 @@ Usage:
webber-cli status
"""
import asyncio
import os
import sys
from pathlib import Path
@@ -23,13 +22,16 @@ from rich.panel import Panel
from rich.prompt import Confirm
from webber_cli.client import WebberClient, PermissionMode
from webber_cli.config import get_config, init_config, get_config_path, CONFIG_FILE
from webber_cli.theme import get_console, get_theme
# === Prompt Toolkit Setup ===
# History file location
HISTORY_FILE = Path.home() / ".webber_history"
def _get_history_file() -> Path:
"""Get history file path from config."""
config = get_config()
return Path(config.history.file).expanduser()
# Built-in commands for completion
BUILTIN_COMMANDS = [
@@ -87,7 +89,7 @@ PROMPT_STYLE = Style.from_dict({
def create_prompt_session(working_dir: str) -> PromptSession:
"""Create a configured prompt session with history and completion."""
return PromptSession(
history=FileHistory(str(HISTORY_FILE)),
history=FileHistory(str(_get_history_file())),
auto_suggest=AutoSuggestFromHistory(),
completer=WebberCompleter(working_dir),
style=PROMPT_STYLE,
@@ -103,12 +105,15 @@ app = typer.Typer(
console = get_console()
# Default API URL (can be overridden via env or option)
# Development port is 8095, production is 8086
DEFAULT_API_URL = os.environ.get("WEBBER_API_URL", "http://localhost:8095")
# Default API key for conversation API (dev mode accepts any non-empty key)
DEFAULT_API_KEY = os.environ.get("WEBBER_API_KEY", "webber-cli-dev-key")
def _get_api_url() -> str:
"""Get API URL from config (with env override already applied)."""
return get_config().api.url
def _get_api_key() -> str:
"""Get API key from config (with env override already applied)."""
return get_config().api.key or "webber-cli-dev-key"
def version_callback(value: bool) -> None:
@@ -154,10 +159,10 @@ def main(
@app.command()
def sessions(
api_url: str = typer.Option(
DEFAULT_API_URL,
None,
"--api",
"-a",
help="Webber API URL",
help="Webber API URL (default from config)",
),
limit: int = typer.Option(
20,
@@ -171,12 +176,13 @@ def sessions(
Shows recent sessions that can be resumed with 'chat --resume <id>'.
"""
asyncio.run(_list_sessions(api_url, limit))
url = api_url or _get_api_url()
asyncio.run(_list_sessions(url, limit))
async def _list_sessions(api_url: str, limit: int) -> None:
"""List conversation sessions."""
async with WebberClient(api_url, api_key=DEFAULT_API_KEY) as client:
async with WebberClient(api_url, api_key=_get_api_key()) as client:
# Check API health
if not await client.health_check():
console.print(f"[error]Error:[/] Cannot connect to Webber API at {api_url}")
@@ -223,16 +229,16 @@ def chat(
help="Working directory for the agent",
),
api_url: str = typer.Option(
DEFAULT_API_URL,
None,
"--api",
"-a",
help="Webber API URL",
help="Webber API URL (default from config)",
),
mode: str = typer.Option(
"default",
None,
"--mode",
"-m",
help="Permission mode: default, plan (read-only), auto_accept (no prompts)",
help="Permission mode: default, plan, auto_accept (default from config)",
),
resume: str = typer.Option(
None,
@@ -241,10 +247,10 @@ def chat(
help="Resume a previous session by ID (use 'sessions' to list)",
),
stream: bool = typer.Option(
True,
None,
"--stream/--no-stream",
"-s",
help="Stream responses in real-time",
help="Stream responses in real-time (default from config)",
),
) -> None:
"""
@@ -263,6 +269,12 @@ def chat(
Use --resume to continue a previous session.
"""
# Resolve config defaults
config = get_config()
url = api_url or config.api.url
mode_str = mode or config.cli.mode
use_stream = stream if stream is not None else config.cli.stream
working_dir = str(Path(directory).resolve())
if not Path(working_dir).exists():
@@ -271,9 +283,9 @@ def chat(
# Parse and validate mode
try:
permission_mode = PermissionMode(mode)
permission_mode = PermissionMode(mode_str)
except ValueError:
console.print(f"[error]Error:[/] Invalid mode: {mode}")
console.print(f"[error]Error:[/] Invalid mode: {mode_str}")
console.print("[dim]Valid modes: default, plan, auto_accept[/]")
raise typer.Exit(1)
@@ -284,7 +296,7 @@ def chat(
permission_mode = PermissionMode.default
try:
asyncio.run(_chat_loop(api_url, working_dir, permission_mode, stream, resume))
asyncio.run(_chat_loop(url, working_dir, permission_mode, use_stream, resume))
except KeyboardInterrupt:
console.print("\n[dim]Goodbye![/]")
@@ -302,7 +314,7 @@ async def _chat_loop(
conversation_id: str | None = None
conversation_title: str | None = None
async with WebberClient(api_url, api_key=DEFAULT_API_KEY) as client:
async with WebberClient(api_url, api_key=_get_api_key()) as client:
# Check API health
if not await client.health_check():
console.print(f"[error]Error:[/] Cannot connect to Webber API at {api_url}")
@@ -510,14 +522,15 @@ async def _chat_loop(
@app.command()
def status(
api_url: str = typer.Option(
DEFAULT_API_URL,
None,
"--api",
"-a",
help="Webber API URL",
help="Webber API URL (default from config)",
),
) -> None:
"""Check API status and list available agents."""
asyncio.run(_status(api_url))
url = api_url or _get_api_url()
asyncio.run(_status(url))
async def _status(api_url: str) -> None:
@@ -536,6 +549,54 @@ async def _status(api_url: str) -> None:
console.print("[error]Status:[/] Cannot connect")
@app.command()
def config(
init: bool = typer.Option(
False,
"--init",
"-i",
help="Initialize config file with defaults",
),
) -> None:
"""
Show or initialize configuration.
Without --init, displays current config and source.
With --init, creates ~/.webber/config.toml with defaults.
"""
if init:
path = init_config()
console.print(f"[success]Config initialized:[/] {path}")
console.print("[dim]Edit this file to customize settings.[/]")
return
# Show current config
cfg = get_config()
config_path = get_config_path()
console.print("[title]Webber Configuration[/]\n")
if config_path:
console.print(f"[dim]Config file:[/] {config_path}")
else:
console.print(f"[dim]Config file:[/] [warning]Not found[/] (using defaults)")
console.print(f"[dim]Run 'webber-cli config --init' to create {CONFIG_FILE}[/]")
console.print()
console.print("[info]API Settings[/]")
console.print(f" url: {cfg.api.url}")
console.print(f" key: {'***' if cfg.api.key else '[dim]not set[/]'}")
console.print()
console.print("[info]CLI Settings[/]")
console.print(f" mode: {cfg.cli.mode}")
console.print(f" stream: {cfg.cli.stream}")
console.print()
console.print("[info]History[/]")
console.print(f" file: {cfg.history.file}")
# Keep 'explore' as an alias for 'chat --mode plan' for backwards compatibility
@app.command(hidden=True)
def explore(
@@ -547,16 +608,16 @@ def explore(
help="Working directory",
),
api_url: str = typer.Option(
DEFAULT_API_URL,
None,
"--api",
"-a",
help="Webber API URL",
help="Webber API URL (default from config)",
),
stream: bool = typer.Option(
True,
None,
"--stream/--no-stream",
"-s",
help="Stream responses in real-time",
help="Stream responses in real-time (default from config)",
),
) -> None:
"""
@@ -564,6 +625,11 @@ def explore(
Runs the Task agent in plan (read-only) mode for a single query.
"""
# Resolve config defaults
config = get_config()
url = api_url or config.api.url
use_stream = stream if stream is not None else config.cli.stream
console.print("[dim]Note: 'explore' is deprecated. Use 'chat --mode plan' for interactive mode.[/]")
console.print()
@@ -573,7 +639,7 @@ def explore(
console.print(f"[error]Error:[/] Directory not found: {working_dir}")
raise typer.Exit(1)
asyncio.run(_explore(api_url, query, working_dir, stream))
asyncio.run(_explore(url, query, working_dir, use_stream))
async def _explore(