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

feat: add permission modes and CLI orchestration layer

Browse Source 
Permission Modes:
- Add default/plan/auto_accept modes controlling tool access
- Plan mode restricts Task agent to read-only tools only
- Auto-accept mode bypasses approval prompts (with confirmation)

Approval Scaffolding:
- Add ApprovalRule/ApprovalRuleSet for granular tool control
- Pattern-based matching on tool name and arguments
- Default rules for common safe/dangerous patterns
- Prep for future bidirectional approval flow

CLI Refactor:
- Default to Task agent (main orchestrator)
- Add --mode flag and runtime mode switching
- Integrate prompt_toolkit for better UX:
  - Persistent command history (~/.webber_history)
  - Tab completion for commands and file paths
  - Auto-suggest from history
- Deprecate standalone 'explore' command

Other:
- Split CHANGELOG.md into per-package files
- Update AGENTS.md release procedure for both packages

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
jpmschweitzer
2026-01-14 08:48:07 +01:00
co-authored by Claude Opus 4.5
parent acf231eb66
commit b7956f88ed
14 changed files with 1047 additions and 367 deletions
Download Patch File Download Diff File Expand all files Collapse all files
webber-cli/webber_cli/client.py
+25 -2
View File
@@ -2,20 +2,36 @@
Webber API client.
Communicates with the Webber API backend for agent execution.
Supports permission modes for controlling agent tool access.
"""
import json
import httpx
from collections.abc import AsyncIterator
from dataclasses import dataclass
from enum import Enum
from typing import Any
class PermissionMode(str, Enum):
"""
Permission modes controlling agent tool access.
- default: All tools available (approval may be required)
- plan: Read-only tools only
- auto_accept: All tools, no approval prompts
"""
default = "default"
plan = "plan"
auto_accept = "auto_accept"
@dataclass
class AgentResponse:
"""Response from agent execution."""
response: str
agent_type: str
success: bool
mode: PermissionMode = PermissionMode.default
error: str | None = None
@@ -104,14 +120,16 @@ class WebberClient:
agent_type: str,
prompt: str,
working_dir: str = ".",
mode: PermissionMode = PermissionMode.default,
) -> AgentResponse:
"""
Run an agent with the given prompt.
Args:
agent_type: Type of agent (e.g., "explore")
agent_type: Type of agent (e.g., "task")
prompt: User prompt/query
working_dir: Working directory for the agent
mode: Permission mode controlling tool access
Returns:
AgentResponse with the result
@@ -123,6 +141,7 @@ class WebberClient:
"agent_type": agent_type,
"prompt": prompt,
"working_dir": working_dir,
"mode": mode.value,
},
)
response.raise_for_status()
@@ -131,6 +150,7 @@ class WebberClient:
response=data.get("response", ""),
agent_type=data.get("agent_type", agent_type),
success=data.get("success", True),
mode=PermissionMode(data.get("mode", "default")),
error=data.get("error"),
)
@@ -139,14 +159,16 @@ class WebberClient:
agent_type: str,
prompt: str,
working_dir: str = ".",
mode: PermissionMode = PermissionMode.default,
) -> AsyncIterator[str]:
"""
Run an agent with streaming response.
Args:
agent_type: Type of agent (e.g., "explore")
agent_type: Type of agent (e.g., "task")
prompt: User prompt/query
working_dir: Working directory for the agent
mode: Permission mode controlling tool access
Yields:
Text chunks as they arrive
@@ -163,6 +185,7 @@ class WebberClient:
"agent_type": agent_type,
"prompt": prompt,
"working_dir": working_dir,
"mode": mode.value,
},
) as response:
response.raise_for_status()
webber-cli/webber_cli/main.py
+273 -110
View File
@@ -5,7 +5,7 @@ Webber CLI - Client for the Webber API.
Usage:
webber-cli --help
webber-cli chat [OPTIONS]
webber-cli explore QUERY [OPTIONS]
webber-cli status
"""
import asyncio
import os
@@ -13,13 +13,87 @@ import sys
from pathlib import Path
import typer
from rich.live import Live
from prompt_toolkit import PromptSession
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from prompt_toolkit.completion import Completer, Completion, PathCompleter
from prompt_toolkit.history import FileHistory
from prompt_toolkit.styles import Style
from rich.markdown import Markdown
from rich.panel import Panel
from rich.prompt import Confirm
from webber_cli.client import WebberClient
from webber_cli.client import WebberClient, PermissionMode
from webber_cli.theme import get_console, get_theme
# === Prompt Toolkit Setup ===
# History file location
HISTORY_FILE = Path.home() / ".webber_history"
# Built-in commands for completion
BUILTIN_COMMANDS = [
"exit",
"quit",
"clear",
"mode plan",
"mode default",
"mode auto_accept",
"cd ",
]
class WebberCompleter(Completer):
"""Custom completer for Webber CLI commands."""
def __init__(self, working_dir: str):
self.working_dir = working_dir
self.path_completer = PathCompleter(expanduser=True)
def get_completions(self, document, complete_event):
text = document.text_before_cursor.lower()
# Complete built-in commands
if not text or not text.startswith("cd "):
for cmd in BUILTIN_COMMANDS:
if cmd.startswith(text):
yield Completion(
cmd,
start_position=-len(text),
display_meta="command",
)
# Complete file paths after "cd "
if text.startswith("cd "):
path_text = text[3:]
# Create a sub-document for path completion
from prompt_toolkit.document import Document
path_doc = Document(path_text, len(path_text))
for completion in self.path_completer.get_completions(path_doc, complete_event):
yield Completion(
"cd " + (path_text + completion.text),
start_position=-len(text),
display_meta="directory",
)
# Prompt style matching Rich theme
PROMPT_STYLE = Style.from_dict({
"prompt": "#5f87d7 bold", # info color
"": "", # default text
})
def create_prompt_session(working_dir: str) -> PromptSession:
"""Create a configured prompt session with history and completion."""
return PromptSession(
history=FileHistory(str(HISTORY_FILE)),
auto_suggest=AutoSuggestFromHistory(),
completer=WebberCompleter(working_dir),
style=PROMPT_STYLE,
complete_while_typing=False, # Only complete on Tab
)
app = typer.Typer(
name="webber-cli",
help="CLI client for the Webber API",
@@ -37,11 +111,28 @@ DEFAULT_API_URL = os.environ.get("WEBBER_API_URL", "http://localhost:8095")
def version_callback(value: bool) -> None:
"""Display version and exit."""
if value:
from cli import __version__
from webber_cli import __version__
console.print(f"[title]webber-cli[/] version [success]{__version__}[/]")
raise typer.Exit()
def _confirm_auto_accept() -> bool:
"""
Prompt user to confirm auto_accept mode.
Returns True if user confirms, False otherwise.
"""
console.print()
console.print("[warning]WARNING:[/] auto_accept mode bypasses all safety prompts.")
console.print("The agent will execute write operations without confirmation.")
console.print()
return Confirm.ask(
"[warning]Are you sure you want to enable auto_accept mode?[/]",
default=False,
console=console,
)
@app.callback()
def main(
version: bool = typer.Option(
@@ -63,7 +154,7 @@ def chat(
".",
"--directory",
"-d",
help="Working directory for exploration",
help="Working directory for the agent",
),
api_url: str = typer.Option(
DEFAULT_API_URL,
@@ -71,10 +162,11 @@ def chat(
"-a",
help="Webber API URL",
),
agent: str = typer.Option(
"explore",
"--agent",
help="Agent to use",
mode: str = typer.Option(
"default",
"--mode",
"-m",
help="Permission mode: default, plan (read-only), auto_accept (no prompts)",
),
stream: bool = typer.Option(
True,
@@ -84,9 +176,18 @@ def chat(
),
) -> None:
"""
Start interactive chat session.
Start interactive chat session with the Task agent.
Connects to the Webber API backend for agent execution.
The Task agent is the main orchestrator that can:
- Explore and analyze codebases
- Plan implementation strategies
- Execute code modifications (in default/auto_accept modes)
- Spawn sub-agents for focused tasks
Permission modes:
- default: Full capabilities with approval prompts for writes
- plan: Read-only mode for safe exploration and planning
- auto_accept: Full capabilities without approval prompts (use with caution)
"""
working_dir = str(Path(directory).resolve())
@@ -94,17 +195,35 @@ def chat(
console.print(f"[error]Error:[/] Directory not found: {working_dir}")
raise typer.Exit(1)
# Parse and validate mode
try:
asyncio.run(_chat_loop(api_url, agent, working_dir, stream))
permission_mode = PermissionMode(mode)
except ValueError:
console.print(f"[error]Error:[/] Invalid mode: {mode}")
console.print("[dim]Valid modes: default, plan, auto_accept[/]")
raise typer.Exit(1)
# Confirm auto_accept mode (security risk)
if permission_mode == PermissionMode.auto_accept:
if not _confirm_auto_accept():
console.print("[dim]Cancelled. Using default mode instead.[/]")
permission_mode = PermissionMode.default
try:
asyncio.run(_chat_loop(api_url, working_dir, permission_mode, stream))
except KeyboardInterrupt:
console.print("\n[dim]Goodbye![/]")
async def _chat_loop(
api_url: str, agent_type: str, working_dir: str, stream: bool = True
api_url: str,
working_dir: str,
mode: PermissionMode,
stream: bool = True,
) -> None:
"""Interactive chat loop."""
"""Interactive chat loop with the Task agent."""
theme = get_theme()
agent_type = "task"
async with WebberClient(api_url) as client:
# Check API health
@@ -116,28 +235,45 @@ async def _chat_loop(
# Get agent info
agent_info = await client.get_agent(agent_type)
if not agent_info:
console.print(f"[error]Error:[/] Unknown agent: {agent_type}")
agents = await client.list_agents()
console.print("[dim]Available agents:[/]")
for a in agents:
console.print(f" - {a.name}: {a.description}")
console.print(f"[error]Error:[/] Task agent not found")
return
# Mode display
mode_display = {
PermissionMode.default: "[info]default[/] (full with approvals)",
PermissionMode.plan: "[success]plan[/] (read-only)",
PermissionMode.auto_accept: "[warning]auto_accept[/] (no prompts)",
}
# Welcome message
console.print()
console.print(f"[title]Webber CLI[/] [dim]→ {api_url}[/]")
console.print(f"[dim]Working in:[/] [path]{working_dir}[/]")
console.print(f"[dim]Agent:[/] {agent_info.name} - {agent_info.description}")
mode = "streaming" if stream else "batch"
console.print(f"[dim]Mode:[/] {mode}")
console.print(f"[dim]Mode:[/] {mode_display[mode]}")
console.print(f"[dim]Streaming:[/] {'enabled' if stream else 'disabled'}")
console.print()
console.print("[dim]Type 'exit' to quit, 'clear' to clear screen.[/]")
console.print("[dim]Commands: 'exit' to quit, 'clear' to clear, 'mode <plan|default|auto_accept>' to switch[/]")
console.print("[dim]Tab for completion, Up/Down for history[/]")
console.print()
current_mode = mode
# Create prompt session with history and completion
session = create_prompt_session(working_dir)
# Chat loop
while True:
try:
user_input = console.input("[prompt]>[/] ").strip()
# Use prompt_toolkit for input (with history and completion)
try:
user_input = await session.prompt_async(
[("class:prompt", "> ")],
)
user_input = user_input.strip()
except EOFError:
# Ctrl+D pressed
console.print("[dim]Goodbye![/]")
break
if not user_input:
continue
@@ -152,33 +288,52 @@ async def _chat_loop(
if user_input.lower().startswith("cd "):
new_dir = user_input[3:].strip()
new_path = Path(new_dir).resolve()
# Handle ~ expansion
new_path = Path(new_dir).expanduser().resolve()
if new_path.exists() and new_path.is_dir():
working_dir = str(new_path)
# Update completer's working directory
session.completer.working_dir = working_dir
console.print(f"[info]Changed to:[/] [path]{working_dir}[/]")
else:
console.print(f"[error]Directory not found:[/] {new_dir}")
continue
# Mode switching
if user_input.lower().startswith("mode "):
new_mode_str = user_input[5:].strip()
try:
new_mode = PermissionMode(new_mode_str)
if new_mode == PermissionMode.auto_accept:
if not _confirm_auto_accept():
console.print("[dim]Mode unchanged.[/]")
continue
current_mode = new_mode
console.print(f"[info]Mode changed to:[/] {mode_display[current_mode]}")
except ValueError:
console.print(f"[error]Invalid mode:[/] {new_mode_str}")
console.print("[dim]Valid modes: default, plan, auto_accept[/]")
continue
console.print()
if stream:
# Stream response in real-time
full_response = ""
try:
async for chunk in client.run_agent_stream(
agent_type, user_input, working_dir
agent_type, user_input, working_dir, current_mode
):
sys.stdout.write(chunk)
sys.stdout.flush()
full_response += chunk
console.print() # Newline after streaming
except Exception as e:
console.print(f"\n[error]Stream error:[/] {e}")
else:
# Batch mode with spinner
with console.status("[info]Thinking...[/]", spinner=theme.spinner):
result = await client.run_agent(agent_type, user_input, working_dir)
result = await client.run_agent(
agent_type, user_input, working_dir, current_mode
)
if result.success:
console.print(Markdown(result.response))
@@ -194,87 +349,6 @@ async def _chat_loop(
console.print(f"[error]Error:[/] {e}")
@app.command()
def explore(
query: str = typer.Argument(..., help="What to search for"),
directory: str = typer.Option(
".",
"--directory",
"-d",
help="Working directory",
),
api_url: str = typer.Option(
DEFAULT_API_URL,
"--api",
"-a",
help="Webber API URL",
),
stream: bool = typer.Option(
True,
"--stream/--no-stream",
"-s",
help="Stream responses in real-time",
),
) -> None:
"""
One-shot codebase exploration.
Sends a query to the Webber API and displays the result.
"""
working_dir = str(Path(directory).resolve())
if not Path(working_dir).exists():
console.print(f"[error]Error:[/] Directory not found: {working_dir}")
raise typer.Exit(1)
asyncio.run(_explore(api_url, query, working_dir, stream))
async def _explore(
api_url: str, query: str, working_dir: str, stream: bool = True
) -> None:
"""Execute exploration query."""
theme = get_theme()
async with WebberClient(api_url) as client:
# Check API health
if not await client.health_check():
console.print(f"[error]Error:[/] Cannot connect to Webber API at {api_url}")
console.print("[dim]Make sure the server is running: ./wakeup.sh[/]")
return
console.print(f"[dim]Exploring:[/] [path]{working_dir}[/]")
console.print(f"[dim]Query:[/] {query}")
console.print()
if stream:
# Stream response in real-time
try:
async for chunk in client.run_agent_stream("explore", query, working_dir):
sys.stdout.write(chunk)
sys.stdout.flush()
console.print() # Newline after streaming
except Exception as e:
console.print(f"\n[error]Stream error:[/] {e}")
else:
# Batch mode with spinner
with console.status("[info]Searching...[/]", spinner=theme.spinner):
result = await client.run_agent("explore", query, working_dir)
if result.success:
console.print(Panel(
Markdown(result.response),
title="[success]Findings[/]",
border_style=theme.colors.border_success,
))
else:
console.print(Panel(
f"[error]{result.error}[/]",
title="[error]Error[/]",
border_style=theme.colors.border_error,
))
@app.command()
def status(
api_url: str = typer.Option(
@@ -304,5 +378,94 @@ async def _status(api_url: str) -> None:
console.print("[error]Status:[/] Cannot connect")
# Keep 'explore' as an alias for 'chat --mode plan' for backwards compatibility
@app.command(hidden=True)
def explore(
query: str = typer.Argument(..., help="What to search for"),
directory: str = typer.Option(
".",
"--directory",
"-d",
help="Working directory",
),
api_url: str = typer.Option(
DEFAULT_API_URL,
"--api",
"-a",
help="Webber API URL",
),
stream: bool = typer.Option(
True,
"--stream/--no-stream",
"-s",
help="Stream responses in real-time",
),
) -> None:
"""
[DEPRECATED] One-shot exploration (use 'chat --mode plan' instead).
Runs the Task agent in plan (read-only) mode for a single query.
"""
console.print("[dim]Note: 'explore' is deprecated. Use 'chat --mode plan' for interactive mode.[/]")
console.print()
working_dir = str(Path(directory).resolve())
if not Path(working_dir).exists():
console.print(f"[error]Error:[/] Directory not found: {working_dir}")
raise typer.Exit(1)
asyncio.run(_explore(api_url, query, working_dir, stream))
async def _explore(
api_url: str, query: str, working_dir: str, stream: bool = True
) -> None:
"""Execute exploration query in plan mode."""
theme = get_theme()
mode = PermissionMode.plan
async with WebberClient(api_url) as client:
# Check API health
if not await client.health_check():
console.print(f"[error]Error:[/] Cannot connect to Webber API at {api_url}")
console.print("[dim]Make sure the server is running: ./wakeup.sh[/]")
return
console.print(f"[dim]Exploring:[/] [path]{working_dir}[/]")
console.print(f"[dim]Query:[/] {query}")
console.print(f"[dim]Mode:[/] [success]plan[/] (read-only)")
console.print()
if stream:
# Stream response in real-time
try:
async for chunk in client.run_agent_stream(
"task", query, working_dir, mode
):
sys.stdout.write(chunk)
sys.stdout.flush()
console.print() # Newline after streaming
except Exception as e:
console.print(f"\n[error]Stream error:[/] {e}")
else:
# Batch mode with spinner
with console.status("[info]Searching...[/]", spinner=theme.spinner):
result = await client.run_agent("task", query, working_dir, mode)
if result.success:
console.print(Panel(
Markdown(result.response),
title="[success]Findings[/]",
border_style=theme.colors.border_success,
))
else:
console.print(Panel(
f"[error]{result.error}[/]",
title="[error]Error[/]",
border_style=theme.colors.border_error,
))
if __name__ == "__main__":
app()