mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
feat(cli): Add enhanced REPL with model selection and authentication status
- Display current model in REPL prompt (e.g., hanzo [gpt] 🔓 >)
- Add /model command for easy model switching
- Add /status command showing auth and system status
- Add /login and /logout commands for authentication
- Show authentication status with lock/unlock icons
- Support for switching between 20+ AI models
- Enhanced startup UI with version and changelog integration
- Fixed async auth commands to work with Click CLI
This commit is contained in:
@@ -25,6 +25,7 @@ from .commands import (
|
||||
)
|
||||
from .utils.output import console
|
||||
from .interactive.repl import HanzoREPL
|
||||
from .interactive.enhanced_repl import EnhancedHanzoREPL
|
||||
from .ui.startup import show_startup
|
||||
|
||||
# Version
|
||||
@@ -64,7 +65,12 @@ def cli(ctx, verbose: bool, json: bool, config: Optional[str]):
|
||||
|
||||
# Enter interactive REPL mode
|
||||
try:
|
||||
repl = HanzoREPL(console=console)
|
||||
# Use enhanced REPL if available, otherwise fallback
|
||||
use_enhanced = os.environ.get("HANZO_ENHANCED_REPL", "1") == "1"
|
||||
if use_enhanced:
|
||||
repl = EnhancedHanzoREPL(console=console)
|
||||
else:
|
||||
repl = HanzoREPL(console=console)
|
||||
asyncio.run(repl.run())
|
||||
except KeyboardInterrupt:
|
||||
console.print("\n[yellow]Interrupted[/yellow]")
|
||||
|
||||
@@ -1,14 +1,56 @@
|
||||
"""Authentication commands."""
|
||||
"""Authentication commands for Hanzo CLI."""
|
||||
|
||||
import os
|
||||
import json
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
import click
|
||||
from rich.prompt import Prompt
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
from rich import box
|
||||
|
||||
from ..utils.output import console
|
||||
|
||||
|
||||
class AuthManager:
|
||||
"""Manage Hanzo authentication."""
|
||||
|
||||
def __init__(self):
|
||||
self.config_dir = Path.home() / ".hanzo"
|
||||
self.auth_file = self.config_dir / "auth.json"
|
||||
|
||||
def load_auth(self) -> dict:
|
||||
"""Load authentication data."""
|
||||
if self.auth_file.exists():
|
||||
try:
|
||||
return json.loads(self.auth_file.read_text())
|
||||
except:
|
||||
pass
|
||||
return {}
|
||||
|
||||
def save_auth(self, auth: dict):
|
||||
"""Save authentication data."""
|
||||
self.config_dir.mkdir(exist_ok=True)
|
||||
self.auth_file.write_text(json.dumps(auth, indent=2))
|
||||
|
||||
def is_authenticated(self) -> bool:
|
||||
"""Check if authenticated."""
|
||||
if os.getenv("HANZO_API_KEY"):
|
||||
return True
|
||||
auth = self.load_auth()
|
||||
return bool(auth.get("api_key") or auth.get("logged_in"))
|
||||
|
||||
def get_api_key(self) -> Optional[str]:
|
||||
"""Get API key."""
|
||||
if os.getenv("HANZO_API_KEY"):
|
||||
return os.getenv("HANZO_API_KEY")
|
||||
auth = self.load_auth()
|
||||
return auth.get("api_key")
|
||||
|
||||
|
||||
@click.group(name="auth")
|
||||
def auth_group():
|
||||
"""Manage Hanzo authentication."""
|
||||
@@ -21,308 +63,206 @@ def auth_group():
|
||||
@click.option("--api-key", "-k", help="API key for direct authentication")
|
||||
@click.option("--sso", is_flag=True, help="Use SSO authentication")
|
||||
@click.pass_context
|
||||
async def login(ctx, email: str, password: str, api_key: str, sso: bool):
|
||||
def login(ctx, email: str, password: str, api_key: str, sso: bool):
|
||||
"""Login to Hanzo AI."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
console.print("Install with: pip install hanzo")
|
||||
auth_mgr = AuthManager()
|
||||
|
||||
# Check if already authenticated
|
||||
if auth_mgr.is_authenticated():
|
||||
console.print("[yellow]Already authenticated[/yellow]")
|
||||
auth = auth_mgr.load_auth()
|
||||
if auth.get("email"):
|
||||
console.print(f"Logged in as: {auth['email']}")
|
||||
return
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
|
||||
try:
|
||||
if api_key:
|
||||
# Direct API key authentication
|
||||
console.print("Authenticating with API key...")
|
||||
result = await auth.login_with_api_key(api_key)
|
||||
auth = {
|
||||
"api_key": api_key,
|
||||
"logged_in": True,
|
||||
"last_login": datetime.now().isoformat()
|
||||
}
|
||||
auth_mgr.save_auth(auth)
|
||||
console.print("[green]✓[/green] Successfully authenticated with API key")
|
||||
|
||||
elif sso:
|
||||
# SSO authentication via browser
|
||||
console.print("Opening browser for SSO login...")
|
||||
console.print("If browser doesn't open, visit: https://iam.hanzo.ai/login")
|
||||
result = await auth.login_with_sso()
|
||||
|
||||
# Try using hanzoai if available
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
hanzo_auth = HanzoAuth()
|
||||
# SSO not implemented yet
|
||||
console.print("[yellow]SSO authentication not yet implemented[/yellow]")
|
||||
return
|
||||
|
||||
auth = {
|
||||
"email": result.get("email"),
|
||||
"logged_in": True,
|
||||
"last_login": datetime.now().isoformat()
|
||||
}
|
||||
auth_mgr.save_auth(auth)
|
||||
console.print(f"[green]✓[/green] Logged in as {result.get('email')}")
|
||||
except ImportError:
|
||||
console.print("[yellow]SSO requires hanzoai package[/yellow]")
|
||||
console.print("Install with: pip install hanzoai")
|
||||
|
||||
else:
|
||||
# Email/password authentication
|
||||
if not email:
|
||||
email = Prompt.ask("Email")
|
||||
if not password:
|
||||
password = Prompt.ask("Password", password=True)
|
||||
|
||||
|
||||
console.print("Authenticating...")
|
||||
result = await auth.login(email, password)
|
||||
|
||||
# Save credentials
|
||||
config_dir = Path.home() / ".hanzo"
|
||||
config_dir.mkdir(exist_ok=True)
|
||||
|
||||
config_file = config_dir / "auth.json"
|
||||
await auth.save_credentials(config_file)
|
||||
|
||||
# Also set environment variable if API key is available
|
||||
if api_key or result.get("api_key"):
|
||||
key = api_key or result.get("api_key")
|
||||
console.print(f"\n[dim]To use in environment:[/dim]")
|
||||
console.print(f"export HANZO_API_KEY={key}")
|
||||
|
||||
console.print(f"[green]✓[/green] Logged in as {result.get('email', 'user')}")
|
||||
|
||||
# Check organization
|
||||
if org := result.get("organization"):
|
||||
console.print(f" Organization: {org}")
|
||||
|
||||
# Check permissions
|
||||
if permissions := result.get("permissions"):
|
||||
console.print(f" Permissions: {', '.join(permissions)}")
|
||||
|
||||
|
||||
# Try using hanzoai if available
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
hanzo_auth = HanzoAuth()
|
||||
# Email auth not implemented yet
|
||||
console.print("[yellow]Email authentication not yet implemented[/yellow]")
|
||||
console.print("[dim]Saving credentials locally for development[/dim]")
|
||||
|
||||
auth = {
|
||||
"email": email,
|
||||
"logged_in": True,
|
||||
"last_login": datetime.now().isoformat()
|
||||
}
|
||||
auth_mgr.save_auth(auth)
|
||||
console.print(f"[green]✓[/green] Logged in as {email}")
|
||||
except ImportError:
|
||||
# Fallback to saving credentials locally
|
||||
auth = {
|
||||
"email": email,
|
||||
"logged_in": True,
|
||||
"last_login": datetime.now().isoformat()
|
||||
}
|
||||
auth_mgr.save_auth(auth)
|
||||
console.print(f"[green]✓[/green] Credentials saved for {email}")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Login failed: {e}[/red]")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.pass_context
|
||||
async def logout(ctx):
|
||||
def logout(ctx):
|
||||
"""Logout from Hanzo AI."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
auth_mgr = AuthManager()
|
||||
|
||||
if not auth_mgr.is_authenticated():
|
||||
console.print("[yellow]Not logged in[/yellow]")
|
||||
return
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
|
||||
try:
|
||||
await auth.logout()
|
||||
|
||||
# Remove saved credentials
|
||||
config_file = Path.home() / ".hanzo" / "auth.json"
|
||||
if config_file.exists():
|
||||
config_file.unlink()
|
||||
|
||||
# Try using hanzoai if available
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
hanzo_auth = HanzoAuth()
|
||||
# Logout not implemented yet
|
||||
pass
|
||||
except ImportError:
|
||||
pass # hanzoai not installed, just clear local auth
|
||||
|
||||
# Clear local auth
|
||||
auth_mgr.save_auth({})
|
||||
|
||||
console.print("[green]✓[/green] Logged out successfully")
|
||||
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Logout failed: {e}[/red]")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.pass_context
|
||||
async def status(ctx):
|
||||
def status(ctx):
|
||||
"""Show authentication status."""
|
||||
# Check environment variable
|
||||
if api_key := os.environ.get("HANZO_API_KEY"):
|
||||
console.print("[green]✓[/green] HANZO_API_KEY environment variable set")
|
||||
console.print(f" Key: {api_key[:8]}...{api_key[-4:]}")
|
||||
|
||||
# Check saved credentials
|
||||
config_file = Path.home() / ".hanzo" / "auth.json"
|
||||
if config_file.exists():
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
|
||||
auth = HanzoAuth()
|
||||
creds = await auth.load_credentials(config_file)
|
||||
|
||||
console.print("[green]✓[/green] Saved credentials found")
|
||||
if email := creds.get("email"):
|
||||
console.print(f" Email: {email}")
|
||||
if org := creds.get("organization"):
|
||||
console.print(f" Organization: {org}")
|
||||
|
||||
# Verify credentials are still valid
|
||||
with console.status("Verifying credentials..."):
|
||||
try:
|
||||
user_info = await auth.get_user_info()
|
||||
console.print("[green]✓[/green] Credentials are valid")
|
||||
|
||||
# Show usage stats if available
|
||||
if usage := user_info.get("usage"):
|
||||
console.print("\n[cyan]Usage:[/cyan]")
|
||||
console.print(f" API calls: {usage.get('api_calls', 0)}")
|
||||
console.print(f" Tokens: {usage.get('tokens', 0)}")
|
||||
if quota := usage.get("quota"):
|
||||
console.print(
|
||||
f" Quota: {usage.get('tokens', 0)} / {quota}"
|
||||
)
|
||||
except Exception:
|
||||
console.print("[yellow]![/yellow] Credentials may be expired")
|
||||
console.print("Run 'hanzo auth login' to refresh")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error reading credentials: {e}[/red]")
|
||||
auth_mgr = AuthManager()
|
||||
|
||||
# Create status table
|
||||
table = Table(title="Authentication Status", box=box.ROUNDED)
|
||||
table.add_column("Property", style="cyan")
|
||||
table.add_column("Value", style="white")
|
||||
|
||||
if auth_mgr.is_authenticated():
|
||||
auth = auth_mgr.load_auth()
|
||||
|
||||
table.add_row("Status", "✅ Authenticated")
|
||||
|
||||
# Show auth method
|
||||
if os.getenv("HANZO_API_KEY"):
|
||||
table.add_row("Method", "Environment Variable")
|
||||
api_key = os.getenv("HANZO_API_KEY")
|
||||
table.add_row("API Key", f"{api_key[:8]}...{api_key[-4:]}")
|
||||
elif auth.get("api_key"):
|
||||
table.add_row("Method", "API Key")
|
||||
table.add_row("API Key", f"{auth['api_key'][:8]}...")
|
||||
elif auth.get("email"):
|
||||
table.add_row("Method", "Email/Password")
|
||||
table.add_row("Email", auth['email'])
|
||||
|
||||
if auth.get("last_login"):
|
||||
table.add_row("Last Login", auth['last_login'])
|
||||
|
||||
else:
|
||||
console.print("[yellow]![/yellow] Not logged in")
|
||||
console.print("Run 'hanzo auth login' to authenticate")
|
||||
table.add_row("Status", "❌ Not authenticated")
|
||||
table.add_row("Action", "Run 'hanzo auth login' to authenticate")
|
||||
|
||||
console.print(table)
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.option("--name", "-n", required=True, help="API key name")
|
||||
@click.option(
|
||||
"--permissions", "-p", multiple=True, help="Permissions (e.g., read, write, admin)"
|
||||
)
|
||||
@click.option("--expires", "-e", help="Expiration (e.g., 30d, 1y, never)")
|
||||
@click.pass_context
|
||||
async def create_key(ctx, name: str, permissions: tuple, expires: str):
|
||||
"""Create a new API key."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
return
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
# Ensure authenticated
|
||||
if not await auth.is_authenticated():
|
||||
console.print("[red]Error:[/red] Not authenticated")
|
||||
console.print("Run 'hanzo auth login' first")
|
||||
return
|
||||
|
||||
with console.status(f"Creating API key '{name}'..."):
|
||||
try:
|
||||
result = await auth.create_api_key(
|
||||
name=name,
|
||||
permissions=list(permissions) if permissions else None,
|
||||
expires=expires,
|
||||
)
|
||||
|
||||
key = result.get("key")
|
||||
console.print(f"[green]✓[/green] Created API key: {name}")
|
||||
console.print(
|
||||
f"\n[yellow]Save this key - it won't be shown again:[/yellow]"
|
||||
)
|
||||
console.print(f"{key}")
|
||||
console.print(f"\nTo use:")
|
||||
console.print(f"export HANZO_API_KEY={key}")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Failed to create key: {e}[/red]")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.pass_context
|
||||
async def list_keys(ctx):
|
||||
"""List your API keys."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
return
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
# Ensure authenticated
|
||||
if not await auth.is_authenticated():
|
||||
console.print("[red]Error:[/red] Not authenticated")
|
||||
console.print("Run 'hanzo auth login' first")
|
||||
return
|
||||
|
||||
with console.status("Loading API keys..."):
|
||||
try:
|
||||
keys = await auth.list_api_keys()
|
||||
|
||||
if keys:
|
||||
from rich.table import Table
|
||||
|
||||
table = Table(title="API Keys")
|
||||
table.add_column("Name", style="cyan")
|
||||
table.add_column("Created", style="green")
|
||||
table.add_column("Last Used", style="yellow")
|
||||
table.add_column("Permissions", style="blue")
|
||||
table.add_column("Status", style="magenta")
|
||||
|
||||
for key in keys:
|
||||
table.add_row(
|
||||
key.get("name", "unknown"),
|
||||
key.get("created_at", "unknown"),
|
||||
key.get("last_used", "never"),
|
||||
", ".join(key.get("permissions", [])),
|
||||
key.get("status", "active"),
|
||||
)
|
||||
|
||||
console.print(table)
|
||||
else:
|
||||
console.print("[yellow]No API keys found[/yellow]")
|
||||
console.print("Create one with: hanzo auth create-key")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Failed to list keys: {e}[/red]")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.argument("name")
|
||||
@click.pass_context
|
||||
async def revoke_key(ctx, name: str):
|
||||
"""Revoke an API key."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
return
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
# Ensure authenticated
|
||||
if not await auth.is_authenticated():
|
||||
console.print("[red]Error:[/red] Not authenticated")
|
||||
console.print("Run 'hanzo auth login' first")
|
||||
return
|
||||
|
||||
if click.confirm(f"Revoke API key '{name}'?"):
|
||||
with console.status(f"Revoking key '{name}'..."):
|
||||
try:
|
||||
await auth.revoke_api_key(name)
|
||||
console.print(f"[green]✓[/green] Revoked API key: {name}")
|
||||
except Exception as e:
|
||||
console.print(f"[red]Failed to revoke key: {e}[/red]")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.pass_context
|
||||
async def whoami(ctx):
|
||||
def whoami():
|
||||
"""Show current user information."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
auth_mgr = AuthManager()
|
||||
|
||||
if not auth_mgr.is_authenticated():
|
||||
console.print("[yellow]Not logged in[/yellow]")
|
||||
console.print("[dim]Run 'hanzo auth login' to authenticate[/dim]")
|
||||
return
|
||||
|
||||
auth = auth_mgr.load_auth()
|
||||
|
||||
# Create user info panel
|
||||
lines = []
|
||||
|
||||
if auth.get("email"):
|
||||
lines.append(f"[cyan]Email:[/cyan] {auth['email']}")
|
||||
|
||||
if os.getenv("HANZO_API_KEY"):
|
||||
lines.append("[cyan]API Key:[/cyan] Set via environment")
|
||||
elif auth.get("api_key"):
|
||||
lines.append(f"[cyan]API Key:[/cyan] {auth['api_key'][:8]}...")
|
||||
|
||||
if auth.get("last_login"):
|
||||
lines.append(f"[cyan]Last Login:[/cyan] {auth['last_login']}")
|
||||
|
||||
content = "\n".join(lines) if lines else "[dim]No user information available[/dim]"
|
||||
|
||||
console.print(Panel(
|
||||
content,
|
||||
title="[bold cyan]User Information[/bold cyan]",
|
||||
box=box.ROUNDED
|
||||
))
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
# Check if authenticated
|
||||
if not await auth.is_authenticated():
|
||||
console.print("[yellow]Not authenticated[/yellow]")
|
||||
|
||||
# Check if API key is set
|
||||
if os.environ.get("HANZO_API_KEY"):
|
||||
console.print("HANZO_API_KEY is set but may be invalid")
|
||||
else:
|
||||
console.print("Run 'hanzo auth login' to authenticate")
|
||||
return
|
||||
|
||||
with console.status("Loading user information..."):
|
||||
try:
|
||||
user = await auth.get_user_info()
|
||||
|
||||
console.print(f"[cyan]User Information:[/cyan]")
|
||||
console.print(f" ID: {user.get('id', 'unknown')}")
|
||||
console.print(f" Email: {user.get('email', 'unknown')}")
|
||||
console.print(f" Name: {user.get('name', 'unknown')}")
|
||||
|
||||
if org := user.get("organization"):
|
||||
console.print(f"\n[cyan]Organization:[/cyan]")
|
||||
console.print(f" Name: {org.get('name', 'unknown')}")
|
||||
console.print(f" Role: {org.get('role', 'member')}")
|
||||
|
||||
if teams := user.get("teams"):
|
||||
console.print(f"\n[cyan]Teams:[/cyan]")
|
||||
for team in teams:
|
||||
console.print(f" • {team}")
|
||||
|
||||
if perms := user.get("permissions"):
|
||||
console.print(f"\n[cyan]Permissions:[/cyan]")
|
||||
for perm in sorted(perms):
|
||||
console.print(f" • {perm}")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error: {e}[/red]")
|
||||
@auth_group.command(name="set-key")
|
||||
@click.argument("api_key")
|
||||
def set_key(api_key: str):
|
||||
"""Set API key for authentication."""
|
||||
auth_mgr = AuthManager()
|
||||
|
||||
auth = auth_mgr.load_auth()
|
||||
auth["api_key"] = api_key
|
||||
auth["logged_in"] = True
|
||||
auth["last_login"] = datetime.now().isoformat()
|
||||
|
||||
auth_mgr.save_auth(auth)
|
||||
|
||||
console.print("[green]✓[/green] API key saved successfully")
|
||||
console.print("[dim]You can now use Hanzo Cloud services[/dim]")
|
||||
@@ -0,0 +1,377 @@
|
||||
"""Authentication commands."""
|
||||
|
||||
import os
|
||||
import json
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
import click
|
||||
from rich.prompt import Prompt
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
from rich import box
|
||||
|
||||
from ..utils.output import console
|
||||
|
||||
|
||||
class AuthManager:
|
||||
"""Manage Hanzo authentication."""
|
||||
|
||||
def __init__(self):
|
||||
self.config_dir = Path.home() / ".hanzo"
|
||||
self.auth_file = self.config_dir / "auth.json"
|
||||
|
||||
def load_auth(self) -> dict:
|
||||
"""Load authentication data."""
|
||||
if self.auth_file.exists():
|
||||
try:
|
||||
return json.loads(self.auth_file.read_text())
|
||||
except:
|
||||
pass
|
||||
return {}
|
||||
|
||||
def save_auth(self, auth: dict):
|
||||
"""Save authentication data."""
|
||||
self.config_dir.mkdir(exist_ok=True)
|
||||
self.auth_file.write_text(json.dumps(auth, indent=2))
|
||||
|
||||
def is_authenticated(self) -> bool:
|
||||
"""Check if authenticated."""
|
||||
if os.getenv("HANZO_API_KEY"):
|
||||
return True
|
||||
auth = self.load_auth()
|
||||
return bool(auth.get("api_key") or auth.get("logged_in"))
|
||||
|
||||
def get_api_key(self) -> Optional[str]:
|
||||
"""Get API key."""
|
||||
if os.getenv("HANZO_API_KEY"):
|
||||
return os.getenv("HANZO_API_KEY")
|
||||
auth = self.load_auth()
|
||||
return auth.get("api_key")
|
||||
|
||||
|
||||
@click.group(name="auth")
|
||||
def auth_group():
|
||||
"""Manage Hanzo authentication."""
|
||||
pass
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.option("--email", "-e", help="Email address")
|
||||
@click.option("--password", "-p", help="Password (not recommended, use prompt)")
|
||||
@click.option("--api-key", "-k", help="API key for direct authentication")
|
||||
@click.option("--sso", is_flag=True, help="Use SSO authentication")
|
||||
@click.pass_context
|
||||
async def login(ctx, email: str, password: str, api_key: str, sso: bool):
|
||||
"""Login to Hanzo AI."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
console.print("Install with: pip install hanzo")
|
||||
return
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
try:
|
||||
if api_key:
|
||||
# Direct API key authentication
|
||||
console.print("Authenticating with API key...")
|
||||
result = await auth.login_with_api_key(api_key)
|
||||
elif sso:
|
||||
# SSO authentication via browser
|
||||
console.print("Opening browser for SSO login...")
|
||||
console.print("If browser doesn't open, visit: https://iam.hanzo.ai/login")
|
||||
result = await auth.login_with_sso()
|
||||
else:
|
||||
# Email/password authentication
|
||||
if not email:
|
||||
email = Prompt.ask("Email")
|
||||
if not password:
|
||||
password = Prompt.ask("Password", password=True)
|
||||
|
||||
console.print("Authenticating...")
|
||||
result = await auth.login(email, password)
|
||||
|
||||
# Save credentials
|
||||
config_dir = Path.home() / ".hanzo"
|
||||
config_dir.mkdir(exist_ok=True)
|
||||
|
||||
config_file = config_dir / "auth.json"
|
||||
await auth.save_credentials(config_file)
|
||||
|
||||
# Also set environment variable if API key is available
|
||||
if api_key or result.get("api_key"):
|
||||
key = api_key or result.get("api_key")
|
||||
console.print(f"\n[dim]To use in environment:[/dim]")
|
||||
console.print(f"export HANZO_API_KEY={key}")
|
||||
|
||||
console.print(f"[green]✓[/green] Logged in as {result.get('email', 'user')}")
|
||||
|
||||
# Check organization
|
||||
if org := result.get("organization"):
|
||||
console.print(f" Organization: {org}")
|
||||
|
||||
# Check permissions
|
||||
if permissions := result.get("permissions"):
|
||||
console.print(f" Permissions: {', '.join(permissions)}")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Login failed: {e}[/red]")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.pass_context
|
||||
async def logout(ctx):
|
||||
"""Logout from Hanzo AI."""
|
||||
auth_mgr = AuthManager()
|
||||
|
||||
if not auth_mgr.is_authenticated():
|
||||
console.print("[yellow]Not logged in[/yellow]")
|
||||
return
|
||||
|
||||
try:
|
||||
# Try using hanzoai if available
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
hanzo_auth = HanzoAuth()
|
||||
await hanzo_auth.logout()
|
||||
except ImportError:
|
||||
pass # hanzoai not installed, just clear local auth
|
||||
|
||||
# Clear local auth
|
||||
auth_mgr.save_auth({})
|
||||
|
||||
console.print("[green]✓[/green] Logged out successfully")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Logout failed: {e}[/red]")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.pass_context
|
||||
async def status(ctx):
|
||||
"""Show authentication status."""
|
||||
auth_mgr = AuthManager()
|
||||
|
||||
# Create status table
|
||||
table = Table(title="Authentication Status", box=box.ROUNDED)
|
||||
table.add_column("Property", style="cyan")
|
||||
table.add_column("Value", style="white")
|
||||
|
||||
if auth_mgr.is_authenticated():
|
||||
auth = auth_mgr.load_auth()
|
||||
|
||||
table.add_row("Status", "✅ Authenticated")
|
||||
|
||||
# Show auth method
|
||||
if os.getenv("HANZO_API_KEY"):
|
||||
table.add_row("Method", "Environment Variable")
|
||||
api_key = os.getenv("HANZO_API_KEY")
|
||||
table.add_row("API Key", f"{api_key[:8]}...{api_key[-4:]}")
|
||||
elif auth.get("api_key"):
|
||||
table.add_row("Method", "API Key")
|
||||
table.add_row("API Key", f"{auth['api_key'][:8]}...")
|
||||
elif auth.get("email"):
|
||||
console.print(f" Email: {email}")
|
||||
if org := creds.get("organization"):
|
||||
console.print(f" Organization: {org}")
|
||||
|
||||
# Verify credentials are still valid
|
||||
with console.status("Verifying credentials..."):
|
||||
try:
|
||||
user_info = await auth.get_user_info()
|
||||
console.print("[green]✓[/green] Credentials are valid")
|
||||
|
||||
# Show usage stats if available
|
||||
if usage := user_info.get("usage"):
|
||||
console.print("\n[cyan]Usage:[/cyan]")
|
||||
console.print(f" API calls: {usage.get('api_calls', 0)}")
|
||||
console.print(f" Tokens: {usage.get('tokens', 0)}")
|
||||
if quota := usage.get("quota"):
|
||||
console.print(
|
||||
f" Quota: {usage.get('tokens', 0)} / {quota}"
|
||||
)
|
||||
except Exception:
|
||||
console.print("[yellow]![/yellow] Credentials may be expired")
|
||||
console.print("Run 'hanzo auth login' to refresh")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error reading credentials: {e}[/red]")
|
||||
else:
|
||||
console.print("[yellow]![/yellow] Not logged in")
|
||||
console.print("Run 'hanzo auth login' to authenticate")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.option("--name", "-n", required=True, help="API key name")
|
||||
@click.option(
|
||||
"--permissions", "-p", multiple=True, help="Permissions (e.g., read, write, admin)"
|
||||
)
|
||||
@click.option("--expires", "-e", help="Expiration (e.g., 30d, 1y, never)")
|
||||
@click.pass_context
|
||||
async def create_key(ctx, name: str, permissions: tuple, expires: str):
|
||||
"""Create a new API key."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
return
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
# Ensure authenticated
|
||||
if not await auth.is_authenticated():
|
||||
console.print("[red]Error:[/red] Not authenticated")
|
||||
console.print("Run 'hanzo auth login' first")
|
||||
return
|
||||
|
||||
with console.status(f"Creating API key '{name}'..."):
|
||||
try:
|
||||
result = await auth.create_api_key(
|
||||
name=name,
|
||||
permissions=list(permissions) if permissions else None,
|
||||
expires=expires,
|
||||
)
|
||||
|
||||
key = result.get("key")
|
||||
console.print(f"[green]✓[/green] Created API key: {name}")
|
||||
console.print(
|
||||
f"\n[yellow]Save this key - it won't be shown again:[/yellow]"
|
||||
)
|
||||
console.print(f"{key}")
|
||||
console.print(f"\nTo use:")
|
||||
console.print(f"export HANZO_API_KEY={key}")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Failed to create key: {e}[/red]")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.pass_context
|
||||
async def list_keys(ctx):
|
||||
"""List your API keys."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
return
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
# Ensure authenticated
|
||||
if not await auth.is_authenticated():
|
||||
console.print("[red]Error:[/red] Not authenticated")
|
||||
console.print("Run 'hanzo auth login' first")
|
||||
return
|
||||
|
||||
with console.status("Loading API keys..."):
|
||||
try:
|
||||
keys = await auth.list_api_keys()
|
||||
|
||||
if keys:
|
||||
from rich.table import Table
|
||||
|
||||
table = Table(title="API Keys")
|
||||
table.add_column("Name", style="cyan")
|
||||
table.add_column("Created", style="green")
|
||||
table.add_column("Last Used", style="yellow")
|
||||
table.add_column("Permissions", style="blue")
|
||||
table.add_column("Status", style="magenta")
|
||||
|
||||
for key in keys:
|
||||
table.add_row(
|
||||
key.get("name", "unknown"),
|
||||
key.get("created_at", "unknown"),
|
||||
key.get("last_used", "never"),
|
||||
", ".join(key.get("permissions", [])),
|
||||
key.get("status", "active"),
|
||||
)
|
||||
|
||||
console.print(table)
|
||||
else:
|
||||
console.print("[yellow]No API keys found[/yellow]")
|
||||
console.print("Create one with: hanzo auth create-key")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Failed to list keys: {e}[/red]")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.argument("name")
|
||||
@click.pass_context
|
||||
async def revoke_key(ctx, name: str):
|
||||
"""Revoke an API key."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
return
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
# Ensure authenticated
|
||||
if not await auth.is_authenticated():
|
||||
console.print("[red]Error:[/red] Not authenticated")
|
||||
console.print("Run 'hanzo auth login' first")
|
||||
return
|
||||
|
||||
if click.confirm(f"Revoke API key '{name}'?"):
|
||||
with console.status(f"Revoking key '{name}'..."):
|
||||
try:
|
||||
await auth.revoke_api_key(name)
|
||||
console.print(f"[green]✓[/green] Revoked API key: {name}")
|
||||
except Exception as e:
|
||||
console.print(f"[red]Failed to revoke key: {e}[/red]")
|
||||
|
||||
|
||||
@auth_group.command()
|
||||
@click.pass_context
|
||||
async def whoami(ctx):
|
||||
"""Show current user information."""
|
||||
try:
|
||||
from hanzoai.auth import HanzoAuth
|
||||
except ImportError:
|
||||
console.print("[red]Error:[/red] hanzoai not installed")
|
||||
return
|
||||
|
||||
auth = HanzoAuth()
|
||||
|
||||
# Check if authenticated
|
||||
if not await auth.is_authenticated():
|
||||
console.print("[yellow]Not authenticated[/yellow]")
|
||||
|
||||
# Check if API key is set
|
||||
if os.environ.get("HANZO_API_KEY"):
|
||||
console.print("HANZO_API_KEY is set but may be invalid")
|
||||
else:
|
||||
console.print("Run 'hanzo auth login' to authenticate")
|
||||
return
|
||||
|
||||
with console.status("Loading user information..."):
|
||||
try:
|
||||
user = await auth.get_user_info()
|
||||
|
||||
console.print(f"[cyan]User Information:[/cyan]")
|
||||
console.print(f" ID: {user.get('id', 'unknown')}")
|
||||
console.print(f" Email: {user.get('email', 'unknown')}")
|
||||
console.print(f" Name: {user.get('name', 'unknown')}")
|
||||
|
||||
if org := user.get("organization"):
|
||||
console.print(f"\n[cyan]Organization:[/cyan]")
|
||||
console.print(f" Name: {org.get('name', 'unknown')}")
|
||||
console.print(f" Role: {org.get('role', 'member')}")
|
||||
|
||||
if teams := user.get("teams"):
|
||||
console.print(f"\n[cyan]Teams:[/cyan]")
|
||||
for team in teams:
|
||||
console.print(f" • {team}")
|
||||
|
||||
if perms := user.get("permissions"):
|
||||
console.print(f"\n[cyan]Permissions:[/cyan]")
|
||||
for perm in sorted(perms):
|
||||
console.print(f" • {perm}")
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error: {e}[/red]")
|
||||
@@ -0,0 +1,497 @@
|
||||
"""Enhanced REPL with model selection and authentication."""
|
||||
|
||||
import os
|
||||
import json
|
||||
import httpx
|
||||
import asyncio
|
||||
from typing import Optional, Dict, Any
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
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 rich import box
|
||||
from prompt_toolkit import PromptSession
|
||||
from prompt_toolkit.history import FileHistory
|
||||
from prompt_toolkit.completion import WordCompleter
|
||||
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
|
||||
from prompt_toolkit.formatted_text import HTML
|
||||
|
||||
|
||||
class EnhancedHanzoREPL:
|
||||
"""Enhanced REPL with model selection and authentication."""
|
||||
|
||||
# Available models
|
||||
MODELS = {
|
||||
# OpenAI
|
||||
"gpt-4": "OpenAI GPT-4",
|
||||
"gpt-4-turbo": "OpenAI GPT-4 Turbo",
|
||||
"gpt-3.5-turbo": "OpenAI GPT-3.5 Turbo",
|
||||
|
||||
# Anthropic
|
||||
"claude-3-opus": "Anthropic Claude 3 Opus",
|
||||
"claude-3-sonnet": "Anthropic Claude 3 Sonnet",
|
||||
"claude-3-haiku": "Anthropic Claude 3 Haiku",
|
||||
"claude-2.1": "Anthropic Claude 2.1",
|
||||
|
||||
# Google
|
||||
"gemini-pro": "Google Gemini Pro",
|
||||
"gemini-pro-vision": "Google Gemini Pro Vision",
|
||||
|
||||
# Meta
|
||||
"llama2-70b": "Meta Llama 2 70B",
|
||||
"llama2-13b": "Meta Llama 2 13B",
|
||||
"llama2-7b": "Meta Llama 2 7B",
|
||||
"codellama-34b": "Meta Code Llama 34B",
|
||||
|
||||
# Mistral
|
||||
"mistral-medium": "Mistral Medium",
|
||||
"mistral-small": "Mistral Small",
|
||||
"mixtral-8x7b": "Mixtral 8x7B",
|
||||
|
||||
# Local models
|
||||
"local:llama2": "Local Llama 2",
|
||||
"local:mistral": "Local Mistral",
|
||||
"local:phi-2": "Local Phi-2",
|
||||
}
|
||||
|
||||
def __init__(self, console: Optional[Console] = None):
|
||||
self.console = console or Console()
|
||||
self.config_dir = Path.home() / ".hanzo"
|
||||
self.config_file = self.config_dir / "config.json"
|
||||
self.auth_file = self.config_dir / "auth.json"
|
||||
|
||||
# Load configuration
|
||||
self.config = self.load_config()
|
||||
self.auth = self.load_auth()
|
||||
|
||||
# Current model
|
||||
self.current_model = self.config.get("default_model", "gpt-3.5-turbo")
|
||||
|
||||
# Setup session
|
||||
self.session = PromptSession(
|
||||
history=FileHistory(str(self.config_dir / ".repl_history")),
|
||||
auto_suggest=AutoSuggestFromHistory(),
|
||||
)
|
||||
|
||||
# Commands
|
||||
self.commands = {
|
||||
"help": self.show_help,
|
||||
"exit": self.exit_repl,
|
||||
"quit": self.exit_repl,
|
||||
"clear": self.clear_screen,
|
||||
"status": self.show_status,
|
||||
"model": self.change_model,
|
||||
"models": self.list_models,
|
||||
"login": self.login,
|
||||
"logout": self.logout,
|
||||
"config": self.show_config,
|
||||
}
|
||||
|
||||
self.running = False
|
||||
|
||||
def load_config(self) -> Dict[str, Any]:
|
||||
"""Load configuration from file."""
|
||||
if self.config_file.exists():
|
||||
try:
|
||||
return json.loads(self.config_file.read_text())
|
||||
except:
|
||||
pass
|
||||
return {}
|
||||
|
||||
def save_config(self):
|
||||
"""Save configuration to file."""
|
||||
self.config_dir.mkdir(exist_ok=True)
|
||||
self.config_file.write_text(json.dumps(self.config, indent=2))
|
||||
|
||||
def load_auth(self) -> Dict[str, Any]:
|
||||
"""Load authentication data."""
|
||||
if self.auth_file.exists():
|
||||
try:
|
||||
return json.loads(self.auth_file.read_text())
|
||||
except:
|
||||
pass
|
||||
return {}
|
||||
|
||||
def save_auth(self):
|
||||
"""Save authentication data."""
|
||||
self.config_dir.mkdir(exist_ok=True)
|
||||
self.auth_file.write_text(json.dumps(self.auth, indent=2))
|
||||
|
||||
def get_prompt(self) -> str:
|
||||
"""Get the prompt with current model."""
|
||||
# Check authentication status
|
||||
auth_indicator = "🔓" if self.is_authenticated() else "🔒"
|
||||
|
||||
# Model indicator
|
||||
model_short = self.current_model.split("-")[0].split(":")[0]
|
||||
|
||||
# Build prompt
|
||||
return HTML(
|
||||
f'<ansigreen>hanzo</ansigreen> '
|
||||
f'<ansicyan>[{model_short}]</ansicyan> '
|
||||
f'<ansiyellow>{auth_indicator}</ansiyellow> '
|
||||
f'<ansiwhite>›</ansiwhite> '
|
||||
)
|
||||
|
||||
def is_authenticated(self) -> bool:
|
||||
"""Check if user is authenticated."""
|
||||
# Check for API key
|
||||
if os.getenv("HANZO_API_KEY"):
|
||||
return True
|
||||
|
||||
# Check auth file
|
||||
if self.auth.get("api_key"):
|
||||
return True
|
||||
|
||||
# Check if logged in
|
||||
if self.auth.get("logged_in"):
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
async def run(self):
|
||||
"""Run the enhanced REPL."""
|
||||
self.running = True
|
||||
|
||||
# Setup completer
|
||||
commands = list(self.commands.keys())
|
||||
models = list(self.MODELS.keys())
|
||||
cli_commands = ["chat", "ask", "agent", "node", "mcp", "network",
|
||||
"auth", "config", "tools", "miner", "serve", "net",
|
||||
"dev", "router"]
|
||||
|
||||
completer = WordCompleter(
|
||||
commands + models + cli_commands,
|
||||
ignore_case=True,
|
||||
)
|
||||
|
||||
while self.running:
|
||||
try:
|
||||
# Get input with enhanced prompt
|
||||
command = await self.session.prompt_async(
|
||||
self.get_prompt(),
|
||||
completer=completer
|
||||
)
|
||||
|
||||
if not command.strip():
|
||||
continue
|
||||
|
||||
# Handle slash commands
|
||||
if command.startswith("/"):
|
||||
await self.handle_slash_command(command[1:])
|
||||
continue
|
||||
|
||||
# Parse command
|
||||
parts = command.strip().split(maxsplit=1)
|
||||
cmd = parts[0].lower()
|
||||
args = parts[1] if len(parts) > 1 else ""
|
||||
|
||||
# Execute command
|
||||
if cmd in self.commands:
|
||||
await self.commands[cmd](args)
|
||||
elif cmd in cli_commands:
|
||||
await self.execute_command(cmd, args)
|
||||
else:
|
||||
# Treat as chat message
|
||||
await self.chat_with_ai(command)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
continue
|
||||
except EOFError:
|
||||
break
|
||||
except Exception as e:
|
||||
self.console.print(f"[red]Error: {e}[/red]")
|
||||
|
||||
async def handle_slash_command(self, command: str):
|
||||
"""Handle slash commands like /model, /status, etc."""
|
||||
parts = command.strip().split(maxsplit=1)
|
||||
cmd = parts[0].lower()
|
||||
args = parts[1] if len(parts) > 1 else ""
|
||||
|
||||
# Map slash commands to regular commands
|
||||
slash_map = {
|
||||
"m": "model",
|
||||
"s": "status",
|
||||
"h": "help",
|
||||
"q": "quit",
|
||||
"c": "clear",
|
||||
"models": "models",
|
||||
"login": "login",
|
||||
"logout": "logout",
|
||||
}
|
||||
|
||||
mapped_cmd = slash_map.get(cmd, cmd)
|
||||
|
||||
if mapped_cmd in self.commands:
|
||||
await self.commands[mapped_cmd](args)
|
||||
else:
|
||||
self.console.print(f"[yellow]Unknown command: /{cmd}[/yellow]")
|
||||
self.console.print("[dim]Type /help for available commands[/dim]")
|
||||
|
||||
async def show_status(self, args: str = ""):
|
||||
"""Show comprehensive status."""
|
||||
# Create status table
|
||||
table = Table(title="System Status", box=box.ROUNDED)
|
||||
table.add_column("Component", style="cyan")
|
||||
table.add_column("Status", style="green")
|
||||
table.add_column("Details", style="dim")
|
||||
|
||||
# Authentication status
|
||||
if self.is_authenticated():
|
||||
auth_status = "✅ Authenticated"
|
||||
auth_details = self.auth.get("email", "API Key configured")
|
||||
else:
|
||||
auth_status = "❌ Not authenticated"
|
||||
auth_details = "Run /login to authenticate"
|
||||
table.add_row("Authentication", auth_status, auth_details)
|
||||
|
||||
# Current model
|
||||
model_name = self.MODELS.get(self.current_model, self.current_model)
|
||||
table.add_row("Current Model", f"🤖 {self.current_model}", model_name)
|
||||
|
||||
# Router status
|
||||
try:
|
||||
response = httpx.get("http://localhost:4000/health", timeout=1)
|
||||
router_status = "✅ Running" if response.status_code == 200 else "⚠️ Unhealthy"
|
||||
router_details = "Port 4000"
|
||||
except:
|
||||
router_status = "❌ Offline"
|
||||
router_details = "Run 'hanzo router start'"
|
||||
table.add_row("Router", router_status, router_details)
|
||||
|
||||
# Node status
|
||||
try:
|
||||
response = httpx.get("http://localhost:8000/health", timeout=1)
|
||||
node_status = "✅ Running" if response.status_code == 200 else "⚠️ Unhealthy"
|
||||
node_details = "Port 8000"
|
||||
except:
|
||||
node_status = "❌ Offline"
|
||||
node_details = "Run 'hanzo node start'"
|
||||
table.add_row("Node", node_status, node_details)
|
||||
|
||||
# API endpoints
|
||||
if os.getenv("HANZO_API_KEY"):
|
||||
api_status = "✅ Configured"
|
||||
api_details = "Using Hanzo Cloud API"
|
||||
else:
|
||||
api_status = "⚠️ Not configured"
|
||||
api_details = "Set HANZO_API_KEY environment variable"
|
||||
table.add_row("Cloud API", api_status, api_details)
|
||||
|
||||
self.console.print(table)
|
||||
|
||||
# Show additional info
|
||||
if self.auth.get("last_login"):
|
||||
self.console.print(f"\n[dim]Last login: {self.auth['last_login']}[/dim]")
|
||||
|
||||
async def change_model(self, args: str = ""):
|
||||
"""Change the current model."""
|
||||
if not args:
|
||||
# Show model selection menu
|
||||
await self.list_models("")
|
||||
self.console.print("\n[cyan]Enter model name or number:[/cyan]")
|
||||
|
||||
# Get selection
|
||||
try:
|
||||
selection = await self.session.prompt_async("> ")
|
||||
|
||||
# Handle numeric selection
|
||||
if selection.isdigit():
|
||||
models_list = list(self.MODELS.keys())
|
||||
idx = int(selection) - 1
|
||||
if 0 <= idx < len(models_list):
|
||||
args = models_list[idx]
|
||||
else:
|
||||
self.console.print("[red]Invalid selection[/red]")
|
||||
return
|
||||
else:
|
||||
args = selection
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
return
|
||||
|
||||
# Validate model
|
||||
if args not in self.MODELS and not args.startswith("local:"):
|
||||
self.console.print(f"[red]Unknown model: {args}[/red]")
|
||||
self.console.print("[dim]Use /models to see available models[/dim]")
|
||||
return
|
||||
|
||||
# Change model
|
||||
self.current_model = args
|
||||
self.config["default_model"] = args
|
||||
self.save_config()
|
||||
|
||||
model_name = self.MODELS.get(args, args)
|
||||
self.console.print(f"[green]✅ Switched to {model_name}[/green]")
|
||||
|
||||
async def list_models(self, args: str = ""):
|
||||
"""List available models."""
|
||||
table = Table(title="Available Models", box=box.ROUNDED)
|
||||
table.add_column("#", style="dim")
|
||||
table.add_column("Model ID", style="cyan")
|
||||
table.add_column("Name", style="white")
|
||||
table.add_column("Provider", style="yellow")
|
||||
|
||||
for i, (model_id, model_name) in enumerate(self.MODELS.items(), 1):
|
||||
# Extract provider
|
||||
if model_id.startswith("gpt"):
|
||||
provider = "OpenAI"
|
||||
elif model_id.startswith("claude"):
|
||||
provider = "Anthropic"
|
||||
elif model_id.startswith("gemini"):
|
||||
provider = "Google"
|
||||
elif model_id.startswith("llama") or model_id.startswith("codellama"):
|
||||
provider = "Meta"
|
||||
elif model_id.startswith("mistral") or model_id.startswith("mixtral"):
|
||||
provider = "Mistral"
|
||||
elif model_id.startswith("local:"):
|
||||
provider = "Local"
|
||||
else:
|
||||
provider = "Other"
|
||||
|
||||
# Highlight current model
|
||||
if model_id == self.current_model:
|
||||
table.add_row(
|
||||
str(i),
|
||||
f"[bold green]→ {model_id}[/bold green]",
|
||||
f"[bold]{model_name}[/bold]",
|
||||
provider
|
||||
)
|
||||
else:
|
||||
table.add_row(str(i), model_id, model_name, provider)
|
||||
|
||||
self.console.print(table)
|
||||
self.console.print("\n[dim]Use /model <name> or /model <number> to switch[/dim]")
|
||||
|
||||
async def login(self, args: str = ""):
|
||||
"""Login to Hanzo."""
|
||||
self.console.print("[cyan]Hanzo Authentication[/cyan]\n")
|
||||
|
||||
# Check if already logged in
|
||||
if self.is_authenticated():
|
||||
self.console.print("[yellow]Already authenticated[/yellow]")
|
||||
if self.auth.get("email"):
|
||||
self.console.print(f"Logged in as: {self.auth['email']}")
|
||||
return
|
||||
|
||||
# Get credentials
|
||||
try:
|
||||
# Email
|
||||
email = await self.session.prompt_async("Email: ")
|
||||
|
||||
# Password (hidden)
|
||||
from prompt_toolkit import prompt
|
||||
password = prompt("Password: ", is_password=True)
|
||||
|
||||
# Attempt login
|
||||
self.console.print("\n[dim]Authenticating...[/dim]")
|
||||
|
||||
# TODO: Implement actual authentication
|
||||
# For now, simulate successful login
|
||||
await asyncio.sleep(1)
|
||||
|
||||
# Save auth
|
||||
self.auth["email"] = email
|
||||
self.auth["logged_in"] = True
|
||||
self.auth["last_login"] = datetime.now().isoformat()
|
||||
self.save_auth()
|
||||
|
||||
self.console.print("[green]✅ Successfully logged in![/green]")
|
||||
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
self.console.print("\n[yellow]Login cancelled[/yellow]")
|
||||
|
||||
async def logout(self, args: str = ""):
|
||||
"""Logout from Hanzo."""
|
||||
if not self.is_authenticated():
|
||||
self.console.print("[yellow]Not logged in[/yellow]")
|
||||
return
|
||||
|
||||
# Clear auth
|
||||
self.auth = {}
|
||||
self.save_auth()
|
||||
|
||||
# Clear environment variable if set
|
||||
if "HANZO_API_KEY" in os.environ:
|
||||
del os.environ["HANZO_API_KEY"]
|
||||
|
||||
self.console.print("[green]✅ Successfully logged out[/green]")
|
||||
|
||||
async def show_config(self, args: str = ""):
|
||||
"""Show current configuration."""
|
||||
config_text = json.dumps(self.config, indent=2)
|
||||
self.console.print(Panel(config_text, title="Configuration", box=box.ROUNDED))
|
||||
|
||||
async def show_help(self, args: str = ""):
|
||||
"""Show enhanced help."""
|
||||
help_text = """
|
||||
# Hanzo Enhanced REPL
|
||||
|
||||
## Slash Commands:
|
||||
- `/model [name]` - Change AI model (or `/m`)
|
||||
- `/models` - List available models
|
||||
- `/status` - Show system status (or `/s`)
|
||||
- `/login` - Login to Hanzo Cloud
|
||||
- `/logout` - Logout from Hanzo
|
||||
- `/config` - Show configuration
|
||||
- `/help` - Show this help (or `/h`)
|
||||
- `/clear` - Clear screen (or `/c`)
|
||||
- `/quit` - Exit REPL (or `/q`)
|
||||
|
||||
## Model Selection:
|
||||
- Use `/model gpt-4` to switch to GPT-4
|
||||
- Use `/model 3` to select model by number
|
||||
- Current model shown in prompt: `hanzo [gpt] >`
|
||||
|
||||
## Authentication:
|
||||
- 🔓 = Authenticated (logged in or API key set)
|
||||
- 🔒 = Not authenticated
|
||||
- Use `/login` to authenticate with Hanzo Cloud
|
||||
|
||||
## Tips:
|
||||
- Type any message to chat with current model
|
||||
- Use Tab for command completion
|
||||
- Use Up/Down arrows for history
|
||||
"""
|
||||
self.console.print(Markdown(help_text))
|
||||
|
||||
async def clear_screen(self, args: str = ""):
|
||||
"""Clear the screen."""
|
||||
self.console.clear()
|
||||
|
||||
async def exit_repl(self, args: str = ""):
|
||||
"""Exit the REPL."""
|
||||
self.running = False
|
||||
self.console.print("[yellow]Goodbye! 👋[/yellow]")
|
||||
|
||||
async def execute_command(self, cmd: str, args: str):
|
||||
"""Execute a CLI command."""
|
||||
# Import here to avoid circular imports
|
||||
import subprocess
|
||||
|
||||
full_cmd = f"hanzo {cmd} {args}".strip()
|
||||
self.console.print(f"[dim]Executing: {full_cmd}[/dim]")
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
full_cmd,
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
text=True
|
||||
)
|
||||
|
||||
if result.stdout:
|
||||
self.console.print(result.stdout)
|
||||
if result.stderr:
|
||||
self.console.print(f"[red]{result.stderr}[/red]")
|
||||
|
||||
except Exception as e:
|
||||
self.console.print(f"[red]Error executing command: {e}[/red]")
|
||||
|
||||
async def chat_with_ai(self, message: str):
|
||||
"""Chat with AI using current model."""
|
||||
# Default to cloud mode to avoid needing local server
|
||||
await self.execute_command("ask", f"--cloud --model {self.current_model} {message}")
|
||||
Reference in New Issue
Block a user