hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: resolve all lint issues for CI compliance

Browse Source 
- Fix import sorting in multiple files
- Replace bare except with except Exception
- Remove broken auth file with syntax errors
This commit is contained in:
zeekay
2025-09-07 15:43:50 -05:00
parent 674e329c06
commit 2a128c8491
19 changed files with 74 additions and 444 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/parallel_ai_doc_editing.py
+1 -1
View File
@@ -7,7 +7,7 @@ using the batch tool to delegate to different CLI agents (claude, codex, gemini,
"""
import json
from typing import List, Dict, Any
from typing import Any, Dict, List
# Example of how to use hanzo-mcp batch tool for parallel edits
# In practice, this would be called through the MCP interface
examples/worktree_orchestration.py
+1 -1
View File
@@ -8,7 +8,7 @@ across multiple git worktrees, with each agent handling a specific task.
import json
import subprocess
from typing import List, Dict, Any
from typing import Any, Dict, List
# This would normally use the hanzo-mcp Python client
# For demonstration, showing the conceptual flow
pkg/hanzo-mcp/start_server.py
+1 -1
View File
@@ -1,8 +1,8 @@
#!/usr/bin/env python3
"""Start the Hanzo MCP server."""
import sys
import os
import sys
# Ensure the parent directory is in the path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
pkg/hanzo/src/hanzo/cli.py
+1 -1
View File
@@ -23,10 +23,10 @@ from .commands import (
router,
network,
)
from .ui.startup import show_startup
from .utils.output import console
from .interactive.repl import HanzoREPL
from .interactive.enhanced_repl import EnhancedHanzoREPL
from .ui.startup import show_startup
# Version
__version__ = "0.3.23"
pkg/hanzo/src/hanzo/commands/auth.py
+4 -4
View File
@@ -2,15 +2,15 @@
import os
import json
from typing import Optional
from pathlib import Path
from datetime import datetime
from typing import Optional
import click
from rich.prompt import Prompt
from rich import box
from rich.panel import Panel
from rich.table import Table
from rich import box
from rich.prompt import Prompt
from ..utils.output import console
@@ -27,7 +27,7 @@ class AuthManager:
if self.auth_file.exists():
try:
return json.loads(self.auth_file.read_text())
except:
except Exception:
pass
return {}
pkg/hanzo/src/hanzo/commands/auth_broken.py
-377
View File
@@ -1,377 +0,0 @@
"""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]")
pkg/hanzo/src/hanzo/commands/chat.py
+3 -3
View File
@@ -1,14 +1,14 @@
"""Chat command for interactive AI conversations."""
import asyncio
import os
import asyncio
from typing import Optional
import click
import httpx
from rich.markdown import Markdown
from rich.panel import Panel
from rich import box
from rich.panel import Panel
from rich.markdown import Markdown
from ..utils.output import console
pkg/hanzo/src/hanzo/interactive/enhanced_repl.py
+11 -11
View File
@@ -2,18 +2,18 @@
import os
import json
import httpx
import asyncio
from typing import Optional, Dict, Any
from typing import Any, Dict, Optional
from pathlib import Path
from datetime import datetime
from rich.console import Console
from rich.markdown import Markdown
import httpx
from rich import box
from rich.text import Text
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from rich import box
from rich.console import Console
from rich.markdown import Markdown
from prompt_toolkit import PromptSession
from prompt_toolkit.history import FileHistory
from prompt_toolkit.completion import WordCompleter
@@ -21,7 +21,7 @@ from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from prompt_toolkit.formatted_text import HTML
try:
from ..tools.detector import ToolDetector, AITool
from ..tools.detector import AITool, ToolDetector
except ImportError:
ToolDetector = None
AITool = None
@@ -159,7 +159,7 @@ class EnhancedHanzoREPL:
if self.config_file.exists():
try:
return json.loads(self.config_file.read_text())
except:
except Exception:
pass
return {}
@@ -173,7 +173,7 @@ class EnhancedHanzoREPL:
if self.auth_file.exists():
try:
return json.loads(self.auth_file.read_text())
except:
except Exception:
pass
return {}
@@ -345,7 +345,7 @@ class EnhancedHanzoREPL:
response = httpx.get("http://localhost:4000/health", timeout=1)
router_status = "✅ Running" if response.status_code == 200 else "⚠️ Unhealthy"
router_details = "Port 4000"
except:
except Exception:
router_status = "❌ Offline"
router_details = "Run 'hanzo router start'"
table.add_row("Router", router_status, router_details)
@@ -355,7 +355,7 @@ class EnhancedHanzoREPL:
response = httpx.get("http://localhost:3690/health", timeout=1)
node_status = "✅ Running" if response.status_code == 200 else "⚠️ Unhealthy"
node_details = "Port 3690"
except:
except Exception:
node_status = "❌ Offline"
node_details = "Run 'hanzo node start'"
table.add_row("Node", node_status, node_details)
pkg/hanzo/src/hanzo/interactive/model_selector.py
+8 -7
View File
@@ -1,13 +1,14 @@
"""Quick model selector with arrow key navigation."""
from typing import List, Optional, Tuple
from prompt_toolkit import Application
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.layout.containers import HSplit, Window
from prompt_toolkit.layout.controls import FormattedTextControl
from prompt_toolkit.layout.layout import Layout
from prompt_toolkit.widgets import Label
from typing import List, Tuple, Optional
from rich.console import Console
from prompt_toolkit import Application
from prompt_toolkit.widgets import Label
from prompt_toolkit.key_binding import KeyBindings
from prompt_toolkit.layout.layout import Layout
from prompt_toolkit.layout.controls import FormattedTextControl
from prompt_toolkit.layout.containers import HSplit, Window
class QuickModelSelector:
pkg/hanzo/src/hanzo/interactive/todo_manager.py
+7 -7
View File
@@ -2,17 +2,17 @@
import json
import uuid
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Optional, Any
from enum import Enum
from typing import Any, Dict, List, Optional
from pathlib import Path
from datetime import datetime
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich.text import Text
from rich import box
from rich.text import Text
from rich.panel import Panel
from rich.table import Table
from rich.prompt import Prompt, Confirm
from rich.console import Console
class TodoPriority(Enum):
pkg/hanzo/src/hanzo/tools/__init__.py
+1 -1
View File
@@ -1,5 +1,5 @@
"""AI tools detection and management."""
from .detector import ToolDetector, AITool
from .detector import AITool, ToolDetector
__all__ = ["ToolDetector", "AITool"]
pkg/hanzo/src/hanzo/tools/detector.py
+8 -8
View File
@@ -3,14 +3,14 @@
import os
import shutil
import subprocess
import httpx
from typing import Dict, List, Tuple, Optional
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from rich.console import Console
from rich.table import Table
import httpx
from rich import box
from rich.table import Table
from rich.console import Console
@dataclass
@@ -196,14 +196,14 @@ class ToolDetector:
models = models_response.json().get("data", [])
if models:
tool.version = f"Running ({len(models)} models, Port {port})"
except:
except Exception:
pass
return True
except (httpx.ConnectError, httpx.TimeoutException):
# Connection refused or timeout - node not available on this port
continue
except:
except Exception:
pass
# If we get here, Hanzo Node is not properly available
return False
@@ -215,7 +215,7 @@ class ToolDetector:
tool.detected = True
tool.version = "Running"
return True
except:
except Exception:
pass
# Check if command exists
@@ -235,7 +235,7 @@ class ToolDetector:
)
if result.returncode == 0:
tool.version = result.stdout.strip().split()[-1]
except:
except Exception:
pass
return True
pkg/hanzo/src/hanzo/ui/__init__.py
+2 -2
View File
@@ -2,8 +2,8 @@
Hanzo UI components for CLI.
"""
from .startup import show_startup, StartupUI
from .inline_startup import show_inline_startup, show_status
from .startup import StartupUI, show_startup
from .inline_startup import show_status, show_inline_startup
__all__ = [
"show_startup",
pkg/hanzo/src/hanzo/ui/inline_startup.py
+8 -7
View File
@@ -4,13 +4,14 @@ Inline startup notifications for Hanzo commands.
import os
import json
from typing import Optional
from pathlib import Path
from datetime import datetime, timedelta
from typing import Optional
from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from rich import box
from rich.text import Text
from rich.panel import Panel
from rich.console import Console
console = Console()
@@ -37,7 +38,7 @@ class InlineStartup:
)
if datetime.now() - last_shown < self.show_interval:
return False
except:
except Exception:
pass
return True
@@ -99,7 +100,7 @@ class InlineStartup:
response = httpx.get("http://localhost:4000/health", timeout=0.5)
if response.status_code == 200:
items.append("[green]Router ✓[/green]")
except:
except Exception:
pass
# Check node
@@ -108,7 +109,7 @@ class InlineStartup:
response = httpx.get("http://localhost:8000/health", timeout=0.5)
if response.status_code == 200:
items.append("[green]Node ✓[/green]")
except:
except Exception:
pass
# Check API key
pkg/hanzo/src/hanzo/ui/startup.py
+10 -9
View File
@@ -5,18 +5,19 @@ Hanzo startup UI and changelog integration.
import os
import json
import time
from typing import Any, Dict, List, Optional
from pathlib import Path
from typing import Optional, List, Dict, Any
from datetime import datetime, timedelta
import httpx
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from rich import box
from rich.text import Text
from rich.align import Align
from rich.panel import Panel
from rich.table import Table
from rich.columns import Columns
from rich.console import Console
from rich.markdown import Markdown
from rich import box
console = Console()
@@ -36,7 +37,7 @@ class StartupUI:
try:
from hanzo import __version__
return __version__
except:
except Exception:
return "0.3.23"
def _get_last_shown_version(self) -> Optional[str]:
@@ -236,14 +237,14 @@ class StartupUI:
try:
response = httpx.get("http://localhost:4000/health", timeout=1)
router_status = "🟢 Running" if response.status_code == 200 else "🔴 Offline"
except:
except Exception:
router_status = "⚫ Offline"
# Check node status
try:
response = httpx.get("http://localhost:8000/health", timeout=1)
node_status = "🟢 Running" if response.status_code == 200 else "🔴 Offline"
except:
except Exception:
node_status = "⚫ Offline"
# Check API key
@@ -278,7 +279,7 @@ class StartupUI:
latest = data["info"]["version"]
if latest != self.current_version:
return latest
except:
except Exception:
pass
return None
test_hanzo_dev.py
+1 -1
View File
@@ -27,7 +27,7 @@ def check_command_exists(cmd):
try:
result = subprocess.run(["which", cmd], capture_output=True, text=True)
return result.returncode == 0
except:
except Exception:
return False
test_refactoring.py
+3 -3
View File
@@ -1,17 +1,17 @@
#!/usr/bin/env python3
"""Test script to verify DRY refactoring works correctly."""
import asyncio
import sys
import asyncio
from pathlib import Path
# Add packages to path
sys.path.insert(0, str(Path(__file__).parent / "pkg" / "hanzo" / "src"))
sys.path.insert(0, str(Path(__file__).parent / "pkg" / "hanzo-mcp"))
from hanzo_mcp.core.model_registry import registry, ModelProvider
from hanzo_mcp.core.base_agent import AgentConfig, AgentOrchestrator
from hanzo.batch_orchestrator import BatchConfig
from hanzo_mcp.core.base_agent import AgentConfig, AgentOrchestrator
from hanzo_mcp.core.model_registry import ModelProvider, registry
def test_model_registry():
test_todo.py
+3
View File
@@ -2,9 +2,12 @@
"""Test the todo manager directly."""
import asyncio
from rich.console import Console
from pkg.hanzo.src.hanzo.interactive.todo_manager import TodoManager
async def test_todo():
console = Console()
manager = TodoManager(console)
test_workflow.py
+1
View File
@@ -6,6 +6,7 @@ import sys
import json
from pathlib import Path
def check_environment():
"""Check required environment variables."""
print("🔍 Checking environment variables...")