mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
feat(cli): Auto-detect AI coding tools with agent as default
- Detect available AI tools: agent, OpenAI Codex, Gemini CLI, Grok, OpenHands, Hanzo Dev - agent is default when available, with intelligent fallback chain - Display detected tools as 'agent: Tool Name' in REPL - Quick model selector with arrow key navigation (press ↓ for menu) - Background task management with /tasks and /kill commands - Tools integrated into /model command for seamless switching - Version 0.3.26 with enhanced AI tool integration
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "hanzo"
|
||||
version = "0.3.25"
|
||||
version = "0.3.26"
|
||||
description = "Hanzo AI - Complete AI Infrastructure Platform with CLI, Router, MCP, and Agent Runtime"
|
||||
authors = [
|
||||
{name = "Hanzo AI", email = "[email protected]"},
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"""Hanzo - Complete AI Infrastructure Platform with CLI, Router, MCP, and Agent Runtime."""
|
||||
|
||||
__version__ = "0.3.25"
|
||||
__version__ = "0.3.26"
|
||||
__all__ = ["main", "cli", "__version__"]
|
||||
|
||||
from .cli import cli, main
|
||||
|
||||
@@ -20,6 +20,18 @@ from prompt_toolkit.completion import WordCompleter
|
||||
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
|
||||
from prompt_toolkit.formatted_text import HTML
|
||||
|
||||
try:
|
||||
from ..tools.detector import ToolDetector, AITool
|
||||
except ImportError:
|
||||
ToolDetector = None
|
||||
AITool = None
|
||||
|
||||
try:
|
||||
from .model_selector import QuickModelSelector, BackgroundTaskManager
|
||||
except ImportError:
|
||||
QuickModelSelector = None
|
||||
BackgroundTaskManager = None
|
||||
|
||||
|
||||
class EnhancedHanzoREPL:
|
||||
"""Enhanced REPL with model selection and authentication."""
|
||||
@@ -57,6 +69,17 @@ class EnhancedHanzoREPL:
|
||||
"local:mistral": "Local Mistral",
|
||||
"local:phi-2": "Local Phi-2",
|
||||
}
|
||||
|
||||
def get_all_models(self):
|
||||
"""Get all available models including detected tools."""
|
||||
models = dict(self.MODELS)
|
||||
|
||||
# Add detected tools as models
|
||||
if self.detected_tools:
|
||||
for tool in self.detected_tools:
|
||||
models[f"tool:{tool.name}"] = f"{tool.display_name} (Tool)"
|
||||
|
||||
return models
|
||||
|
||||
def __init__(self, console: Optional[Console] = None):
|
||||
self.console = console or Console()
|
||||
@@ -68,8 +91,30 @@ class EnhancedHanzoREPL:
|
||||
self.config = self.load_config()
|
||||
self.auth = self.load_auth()
|
||||
|
||||
# Current model
|
||||
self.current_model = self.config.get("default_model", "gpt-3.5-turbo")
|
||||
# Initialize tool detector
|
||||
self.tool_detector = ToolDetector(console) if ToolDetector else None
|
||||
self.detected_tools = []
|
||||
self.current_tool = None
|
||||
|
||||
# Initialize background task manager
|
||||
self.task_manager = BackgroundTaskManager(console) if BackgroundTaskManager else None
|
||||
|
||||
# Detect available tools and set default
|
||||
if self.tool_detector:
|
||||
self.detected_tools = self.tool_detector.detect_all()
|
||||
default_tool = self.tool_detector.get_default_tool()
|
||||
|
||||
# If Claude Code is available, use it as default
|
||||
if default_tool:
|
||||
self.current_model = f"tool:{default_tool.name}"
|
||||
self.current_tool = default_tool
|
||||
self.console.print(f"[green]✓ Detected {default_tool.display_name} as default AI assistant[/green]")
|
||||
else:
|
||||
# Fallback to regular models
|
||||
self.current_model = self.config.get("default_model", "gpt-3.5-turbo")
|
||||
else:
|
||||
# No tool detector, use regular models
|
||||
self.current_model = self.config.get("default_model", "gpt-3.5-turbo")
|
||||
|
||||
# Setup session
|
||||
self.session = PromptSession(
|
||||
@@ -86,9 +131,14 @@ class EnhancedHanzoREPL:
|
||||
"status": self.show_status,
|
||||
"model": self.change_model,
|
||||
"models": self.list_models,
|
||||
"tools": self.list_tools,
|
||||
"agents": self.list_tools, # Alias for tools
|
||||
"login": self.login,
|
||||
"logout": self.logout,
|
||||
"config": self.show_config,
|
||||
"tasks": self.show_tasks,
|
||||
"kill": self.kill_task,
|
||||
"quick": self.quick_model_select,
|
||||
}
|
||||
|
||||
self.running = False
|
||||
@@ -144,8 +194,17 @@ class EnhancedHanzoREPL:
|
||||
|
||||
def get_model_info(self):
|
||||
"""Get current model info string."""
|
||||
# Determine provider from model name
|
||||
model = self.current_model
|
||||
|
||||
# Check if using a tool
|
||||
if model.startswith("tool:"):
|
||||
if self.current_tool:
|
||||
return f"[dim cyan]agent: {self.current_tool.display_name}[/dim cyan]"
|
||||
else:
|
||||
tool_name = model.replace("tool:", "")
|
||||
return f"[dim cyan]agent: {tool_name}[/dim cyan]"
|
||||
|
||||
# Determine provider from model name
|
||||
if model.startswith("gpt"):
|
||||
provider = "openai"
|
||||
elif model.startswith("claude"):
|
||||
@@ -187,7 +246,8 @@ class EnhancedHanzoREPL:
|
||||
# Get input with simple prompt
|
||||
command = await self.session.prompt_async(
|
||||
self.get_prompt(),
|
||||
completer=completer
|
||||
completer=completer,
|
||||
vi_mode=True # Enable vi mode for better navigation
|
||||
)
|
||||
|
||||
if not command.strip():
|
||||
@@ -302,11 +362,11 @@ class EnhancedHanzoREPL:
|
||||
self.console.print(f"\n[dim]Last login: {self.auth['last_login']}[/dim]")
|
||||
|
||||
async def change_model(self, args: str = ""):
|
||||
"""Change the current model."""
|
||||
"""Change the current model or tool."""
|
||||
if not args:
|
||||
# Show model selection menu
|
||||
await self.list_models("")
|
||||
self.console.print("\n[cyan]Enter model name or number:[/cyan]")
|
||||
self.console.print("\n[cyan]Enter model/tool name or number:[/cyan]")
|
||||
|
||||
# Get selection
|
||||
try:
|
||||
@@ -314,41 +374,87 @@ class EnhancedHanzoREPL:
|
||||
|
||||
# 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]
|
||||
num = int(selection)
|
||||
|
||||
# Check if it's a tool selection
|
||||
if self.detected_tools and num <= len(self.detected_tools):
|
||||
tool = self.detected_tools[num - 1]
|
||||
args = f"tool:{tool.name}"
|
||||
else:
|
||||
self.console.print("[red]Invalid selection[/red]")
|
||||
return
|
||||
# It's a model selection
|
||||
model_idx = num - len(self.detected_tools) - 1 if self.detected_tools else num - 1
|
||||
models_list = list(self.MODELS.keys())
|
||||
if 0 <= model_idx < len(models_list):
|
||||
args = models_list[model_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
|
||||
# Check if it's a tool
|
||||
if args.startswith("tool:") or args in [t.name for t in self.detected_tools] if self.detected_tools else False:
|
||||
# Handle tool selection
|
||||
tool_name = args.replace("tool:", "") if args.startswith("tool:") else args
|
||||
|
||||
# Find the tool
|
||||
tool = None
|
||||
for t in self.detected_tools:
|
||||
if t.name == tool_name or t.display_name.lower() == tool_name.lower():
|
||||
tool = t
|
||||
break
|
||||
|
||||
if tool:
|
||||
self.current_model = f"tool:{tool.name}"
|
||||
self.current_tool = tool
|
||||
self.config["default_model"] = self.current_model
|
||||
self.save_config()
|
||||
self.console.print(f"[green]✅ Switched to {tool.display_name}[/green]")
|
||||
else:
|
||||
self.console.print(f"[red]Tool not found: {tool_name}[/red]")
|
||||
self.console.print("[dim]Use /tools to see available tools[/dim]")
|
||||
|
||||
# Change model
|
||||
self.current_model = args
|
||||
self.config["default_model"] = args
|
||||
self.save_config()
|
||||
# Regular model
|
||||
elif args in self.MODELS or args.startswith("local:"):
|
||||
self.current_model = args
|
||||
self.current_tool = None
|
||||
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]")
|
||||
|
||||
model_name = self.MODELS.get(args, args)
|
||||
self.console.print(f"[green]✅ Switched to {model_name}[/green]")
|
||||
else:
|
||||
self.console.print(f"[red]Unknown model or tool: {args}[/red]")
|
||||
self.console.print("[dim]Use /models or /tools to see available options[/dim]")
|
||||
|
||||
async def list_tools(self, args: str = ""):
|
||||
"""List available AI tools."""
|
||||
if self.tool_detector:
|
||||
self.tool_detector.show_available_tools()
|
||||
else:
|
||||
self.console.print("[yellow]Tool detection not available[/yellow]")
|
||||
|
||||
async def list_models(self, args: str = ""):
|
||||
"""List available models."""
|
||||
table = Table(title="Available Models", box=box.ROUNDED)
|
||||
# Show tools first if available
|
||||
if self.detected_tools:
|
||||
self.console.print("[bold cyan]AI Coding Assistants (Detected):[/bold cyan]")
|
||||
for i, tool in enumerate(self.detected_tools, 1):
|
||||
marker = "→" if self.current_model == f"tool:{tool.name}" else " "
|
||||
self.console.print(f" {marker} {i}. {tool.display_name} ({tool.provider})")
|
||||
self.console.print()
|
||||
|
||||
table = Table(title="Language 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):
|
||||
start_idx = len(self.detected_tools) + 1 if self.detected_tools else 1
|
||||
for i, (model_id, model_name) in enumerate(self.MODELS.items(), start_idx):
|
||||
# Extract provider
|
||||
if model_id.startswith("gpt"):
|
||||
provider = "OpenAI"
|
||||
@@ -446,6 +552,10 @@ class EnhancedHanzoREPL:
|
||||
## Slash Commands:
|
||||
- `/model [name]` - Change AI model (or `/m`)
|
||||
- `/models` - List available models
|
||||
- `/tools` - List available AI tools
|
||||
- `/quick` - Quick model selector (arrow keys)
|
||||
- `/tasks` - Show background tasks
|
||||
- `/kill [id]` - Kill background task
|
||||
- `/status` - Show system status (or `/s`)
|
||||
- `/login` - Login to Hanzo Cloud
|
||||
- `/logout` - Logout from Hanzo
|
||||
@@ -454,6 +564,11 @@ class EnhancedHanzoREPL:
|
||||
- `/clear` - Clear screen (or `/c`)
|
||||
- `/quit` - Exit REPL (or `/q`)
|
||||
|
||||
## Quick Model Selection:
|
||||
- Press ↓ arrow key for quick model selector
|
||||
- Use ↑/↓ to navigate, Enter to select
|
||||
- Esc to cancel
|
||||
|
||||
## Model Selection:
|
||||
- Use `/model gpt-4` to switch to GPT-4
|
||||
- Use `/model 3` to select model by number
|
||||
@@ -505,6 +620,69 @@ class EnhancedHanzoREPL:
|
||||
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}")
|
||||
"""Chat with AI using current model or tool."""
|
||||
# Check if using a tool
|
||||
if self.current_model.startswith("tool:") and self.current_tool:
|
||||
# Use the detected tool directly
|
||||
self.console.print(f"[dim]Using {self.current_tool.display_name}...[/dim]")
|
||||
|
||||
success, output = self.tool_detector.execute_with_tool(self.current_tool, message)
|
||||
|
||||
if success:
|
||||
self.console.print(output)
|
||||
else:
|
||||
# Fallback to regular model
|
||||
self.console.print(f"[yellow]{self.current_tool.display_name} failed, trying cloud model...[/yellow]")
|
||||
await self.execute_command("ask", f"--cloud --model gpt-3.5-turbo {message}")
|
||||
else:
|
||||
# Use regular model through hanzo ask
|
||||
await self.execute_command("ask", f"--cloud --model {self.current_model} {message}")
|
||||
|
||||
async def quick_model_select(self, args: str = ""):
|
||||
"""Quick model selector with arrow keys."""
|
||||
if not QuickModelSelector:
|
||||
self.console.print("[yellow]Quick selector not available[/yellow]")
|
||||
return
|
||||
|
||||
# Prepare tools and models
|
||||
tools = [(f"tool:{t.name}", t.display_name) for t in self.detected_tools] if self.detected_tools else []
|
||||
models = list(self.MODELS.items())
|
||||
|
||||
selector = QuickModelSelector(models, tools, self.current_model)
|
||||
selected = await selector.run()
|
||||
|
||||
if selected:
|
||||
# Change to selected model
|
||||
await self.change_model(selected)
|
||||
|
||||
async def show_tasks(self, args: str = ""):
|
||||
"""Show background tasks."""
|
||||
if self.task_manager:
|
||||
self.task_manager.list_tasks()
|
||||
else:
|
||||
self.console.print("[yellow]Task manager not available[/yellow]")
|
||||
|
||||
async def kill_task(self, args: str = ""):
|
||||
"""Kill a background task."""
|
||||
if not self.task_manager:
|
||||
self.console.print("[yellow]Task manager not available[/yellow]")
|
||||
return
|
||||
|
||||
if args:
|
||||
if args.lower() == "all":
|
||||
self.task_manager.kill_all()
|
||||
else:
|
||||
self.task_manager.kill_task(args)
|
||||
else:
|
||||
# Show tasks and prompt for selection
|
||||
self.task_manager.list_tasks()
|
||||
self.console.print("\n[cyan]Enter task ID to kill (or 'all' for all tasks):[/cyan]")
|
||||
try:
|
||||
task_id = await self.session.prompt_async("> ")
|
||||
if task_id:
|
||||
if task_id.lower() == "all":
|
||||
self.task_manager.kill_all()
|
||||
else:
|
||||
self.task_manager.kill_task(task_id)
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
pass
|
||||
@@ -0,0 +1,166 @@
|
||||
"""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 rich.console import Console
|
||||
|
||||
|
||||
class QuickModelSelector:
|
||||
"""Quick model selector with arrow navigation."""
|
||||
|
||||
def __init__(self, models: List[Tuple[str, str]], tools: List[Tuple[str, str]], current: str):
|
||||
self.models = models
|
||||
self.tools = tools
|
||||
self.current = current
|
||||
self.all_items = tools + models # Tools first, then models
|
||||
self.selected_index = 0
|
||||
|
||||
# Find current selection
|
||||
for i, (item_id, _) in enumerate(self.all_items):
|
||||
if item_id == current:
|
||||
self.selected_index = i
|
||||
break
|
||||
|
||||
def get_display_lines(self) -> List[str]:
|
||||
"""Get display lines for the selector."""
|
||||
lines = []
|
||||
|
||||
if self.tools:
|
||||
lines.append("AI Coding Assistants:")
|
||||
for i, (tool_id, tool_name) in enumerate(self.tools):
|
||||
marker = "→ " if i == self.selected_index else " "
|
||||
lines.append(f"{marker}{tool_name}")
|
||||
|
||||
if self.models:
|
||||
if self.tools:
|
||||
lines.append("") # Empty line
|
||||
lines.append("Language Models:")
|
||||
|
||||
tool_count = len(self.tools)
|
||||
for i, (model_id, model_name) in enumerate(self.models):
|
||||
actual_idx = tool_count + i
|
||||
marker = "→ " if actual_idx == self.selected_index else " "
|
||||
lines.append(f"{marker}{model_name}")
|
||||
|
||||
return lines
|
||||
|
||||
def move_up(self):
|
||||
"""Move selection up."""
|
||||
if self.selected_index > 0:
|
||||
self.selected_index -= 1
|
||||
|
||||
def move_down(self):
|
||||
"""Move selection down."""
|
||||
if self.selected_index < len(self.all_items) - 1:
|
||||
self.selected_index += 1
|
||||
|
||||
def get_selected(self) -> Tuple[str, str]:
|
||||
"""Get the selected item."""
|
||||
if 0 <= self.selected_index < len(self.all_items):
|
||||
return self.all_items[self.selected_index]
|
||||
return None, None
|
||||
|
||||
async def run(self) -> Optional[str]:
|
||||
"""Run the selector and return selected model/tool ID."""
|
||||
kb = KeyBindings()
|
||||
|
||||
@kb.add('up')
|
||||
def _(event):
|
||||
self.move_up()
|
||||
event.app.invalidate()
|
||||
|
||||
@kb.add('down')
|
||||
def _(event):
|
||||
self.move_down()
|
||||
event.app.invalidate()
|
||||
|
||||
@kb.add('enter')
|
||||
def _(event):
|
||||
event.app.exit(result=self.get_selected()[0])
|
||||
|
||||
@kb.add('c-c')
|
||||
@kb.add('escape')
|
||||
def _(event):
|
||||
event.app.exit(result=None)
|
||||
|
||||
def get_text():
|
||||
lines = self.get_display_lines()
|
||||
lines.append("")
|
||||
lines.append("↑/↓: Navigate Enter: Select Esc: Cancel")
|
||||
return "\n".join(lines)
|
||||
|
||||
layout = Layout(
|
||||
Window(
|
||||
FormattedTextControl(get_text),
|
||||
wrap_lines=False
|
||||
)
|
||||
)
|
||||
|
||||
app = Application(
|
||||
layout=layout,
|
||||
key_bindings=kb,
|
||||
full_screen=False,
|
||||
mouse_support=True
|
||||
)
|
||||
|
||||
return await app.run_async()
|
||||
|
||||
|
||||
class BackgroundTaskManager:
|
||||
"""Manage background tasks."""
|
||||
|
||||
def __init__(self, console: Optional[Console] = None):
|
||||
self.console = console or Console()
|
||||
self.tasks = {} # task_id -> process
|
||||
self.next_id = 1
|
||||
|
||||
def add_task(self, name: str, process):
|
||||
"""Add a background task."""
|
||||
task_id = f"task_{self.next_id}"
|
||||
self.next_id += 1
|
||||
self.tasks[task_id] = {
|
||||
"name": name,
|
||||
"process": process,
|
||||
"started": True
|
||||
}
|
||||
return task_id
|
||||
|
||||
def list_tasks(self):
|
||||
"""List all background tasks."""
|
||||
if not self.tasks:
|
||||
self.console.print("[dim]No background tasks running[/dim]")
|
||||
return
|
||||
|
||||
self.console.print("[bold]Background Tasks:[/bold]")
|
||||
for task_id, task in self.tasks.items():
|
||||
status = "🟢 Running" if task["process"].poll() is None else "🔴 Stopped"
|
||||
self.console.print(f" {task_id}: {task['name']} - {status}")
|
||||
|
||||
def kill_task(self, task_id: str):
|
||||
"""Kill a background task."""
|
||||
if task_id in self.tasks:
|
||||
task = self.tasks[task_id]
|
||||
if task["process"].poll() is None:
|
||||
task["process"].terminate()
|
||||
self.console.print(f"[yellow]Terminated {task_id}: {task['name']}[/yellow]")
|
||||
else:
|
||||
self.console.print(f"[dim]Task {task_id} already stopped[/dim]")
|
||||
del self.tasks[task_id]
|
||||
else:
|
||||
self.console.print(f"[red]Task {task_id} not found[/red]")
|
||||
|
||||
def kill_all(self):
|
||||
"""Kill all background tasks."""
|
||||
if not self.tasks:
|
||||
self.console.print("[dim]No tasks to kill[/dim]")
|
||||
return
|
||||
|
||||
for task_id in list(self.tasks.keys()):
|
||||
self.kill_task(task_id)
|
||||
|
||||
self.console.print("[green]All tasks terminated[/green]")
|
||||
@@ -0,0 +1,5 @@
|
||||
"""AI tools detection and management."""
|
||||
|
||||
from .detector import ToolDetector, AITool
|
||||
|
||||
__all__ = ["ToolDetector", "AITool"]
|
||||
@@ -0,0 +1,291 @@
|
||||
"""Detect available AI coding tools and assistants."""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
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
|
||||
from rich import box
|
||||
|
||||
|
||||
@dataclass
|
||||
class AITool:
|
||||
"""Represents an AI coding tool."""
|
||||
name: str
|
||||
command: str
|
||||
display_name: str
|
||||
provider: str
|
||||
priority: int # Lower is higher priority
|
||||
check_command: Optional[str] = None
|
||||
env_var: Optional[str] = None
|
||||
api_endpoint: Optional[str] = None
|
||||
detected: bool = False
|
||||
version: Optional[str] = None
|
||||
path: Optional[str] = None
|
||||
|
||||
|
||||
class ToolDetector:
|
||||
"""Detect and manage available AI coding tools."""
|
||||
|
||||
# Define available tools with priority order
|
||||
TOOLS = [
|
||||
AITool(
|
||||
name="claude-code",
|
||||
command="claude",
|
||||
display_name="Claude Code",
|
||||
provider="anthropic",
|
||||
priority=1,
|
||||
check_command="claude --version",
|
||||
env_var="ANTHROPIC_API_KEY"
|
||||
),
|
||||
AITool(
|
||||
name="hanzo-dev",
|
||||
command="hanzo dev",
|
||||
display_name="Hanzo Dev (Native)",
|
||||
provider="hanzo",
|
||||
priority=2,
|
||||
check_command="hanzo --version",
|
||||
env_var="HANZO_API_KEY"
|
||||
),
|
||||
AITool(
|
||||
name="openai-codex",
|
||||
command="openai",
|
||||
display_name="OpenAI Codex",
|
||||
provider="openai",
|
||||
priority=3,
|
||||
check_command="openai --version",
|
||||
env_var="OPENAI_API_KEY"
|
||||
),
|
||||
AITool(
|
||||
name="gemini-cli",
|
||||
command="gemini",
|
||||
display_name="Gemini CLI",
|
||||
provider="google",
|
||||
priority=4,
|
||||
check_command="gemini --version",
|
||||
env_var="GEMINI_API_KEY"
|
||||
),
|
||||
AITool(
|
||||
name="grok-cli",
|
||||
command="grok",
|
||||
display_name="Grok CLI",
|
||||
provider="xai",
|
||||
priority=5,
|
||||
check_command="grok --version",
|
||||
env_var="GROK_API_KEY"
|
||||
),
|
||||
AITool(
|
||||
name="openhands",
|
||||
command="openhands",
|
||||
display_name="OpenHands CLI",
|
||||
provider="openhands",
|
||||
priority=6,
|
||||
check_command="openhands --version",
|
||||
env_var=None
|
||||
),
|
||||
AITool(
|
||||
name="cursor",
|
||||
command="cursor",
|
||||
display_name="Cursor AI",
|
||||
provider="cursor",
|
||||
priority=7,
|
||||
check_command="cursor --version",
|
||||
env_var=None
|
||||
),
|
||||
AITool(
|
||||
name="codeium",
|
||||
command="codeium",
|
||||
display_name="Codeium",
|
||||
provider="codeium",
|
||||
priority=8,
|
||||
check_command="codeium --version",
|
||||
env_var="CODEIUM_API_KEY"
|
||||
),
|
||||
AITool(
|
||||
name="aider",
|
||||
command="aider",
|
||||
display_name="Aider",
|
||||
provider="aider",
|
||||
priority=9,
|
||||
check_command="aider --version",
|
||||
env_var=None
|
||||
),
|
||||
AITool(
|
||||
name="continue",
|
||||
command="continue",
|
||||
display_name="Continue Dev",
|
||||
provider="continue",
|
||||
priority=10,
|
||||
check_command="continue --version",
|
||||
env_var=None
|
||||
)
|
||||
]
|
||||
|
||||
def __init__(self, console: Optional[Console] = None):
|
||||
self.console = console or Console()
|
||||
self.detected_tools: List[AITool] = []
|
||||
|
||||
def detect_all(self) -> List[AITool]:
|
||||
"""Detect all available AI tools."""
|
||||
self.detected_tools = []
|
||||
|
||||
for tool in self.TOOLS:
|
||||
if self.detect_tool(tool):
|
||||
self.detected_tools.append(tool)
|
||||
|
||||
# Sort by priority
|
||||
self.detected_tools.sort(key=lambda t: t.priority)
|
||||
return self.detected_tools
|
||||
|
||||
def detect_tool(self, tool: AITool) -> bool:
|
||||
"""Detect if a specific tool is available."""
|
||||
# Check if command exists
|
||||
tool.path = shutil.which(tool.command.split()[0])
|
||||
if tool.path:
|
||||
tool.detected = True
|
||||
|
||||
# Try to get version
|
||||
if tool.check_command:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
tool.check_command.split(),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=2
|
||||
)
|
||||
if result.returncode == 0:
|
||||
tool.version = result.stdout.strip().split()[-1]
|
||||
except:
|
||||
pass
|
||||
|
||||
return True
|
||||
|
||||
# Check environment variable as fallback
|
||||
if tool.env_var and os.getenv(tool.env_var):
|
||||
tool.detected = True
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def get_default_tool(self) -> Optional[AITool]:
|
||||
"""Get the default tool based on priority and availability."""
|
||||
if not self.detected_tools:
|
||||
self.detect_all()
|
||||
|
||||
if self.detected_tools:
|
||||
return self.detected_tools[0]
|
||||
return None
|
||||
|
||||
def get_tool_by_name(self, name: str) -> Optional[AITool]:
|
||||
"""Get a specific tool by name."""
|
||||
for tool in self.TOOLS:
|
||||
if tool.name == name or tool.display_name.lower() == name.lower():
|
||||
if self.detect_tool(tool):
|
||||
return tool
|
||||
return None
|
||||
|
||||
def show_available_tools(self):
|
||||
"""Display available tools in a table."""
|
||||
self.detect_all()
|
||||
|
||||
table = Table(title="Available AI Coding Tools", box=box.ROUNDED)
|
||||
table.add_column("#", style="dim")
|
||||
table.add_column("Tool", style="cyan")
|
||||
table.add_column("Provider", style="yellow")
|
||||
table.add_column("Status", style="green")
|
||||
table.add_column("Version", style="blue")
|
||||
table.add_column("Priority", style="magenta")
|
||||
|
||||
for i, tool in enumerate(self.TOOLS, 1):
|
||||
status = "✅ Available" if tool.detected else "❌ Not Found"
|
||||
version = tool.version or "Unknown" if tool.detected else "-"
|
||||
|
||||
# Highlight the default tool
|
||||
if tool.detected and tool == self.detected_tools[0] if self.detected_tools else False:
|
||||
table.add_row(
|
||||
str(i),
|
||||
f"[bold green]→ {tool.display_name}[/bold green]",
|
||||
tool.provider,
|
||||
status,
|
||||
version,
|
||||
str(tool.priority)
|
||||
)
|
||||
else:
|
||||
table.add_row(
|
||||
str(i),
|
||||
tool.display_name,
|
||||
tool.provider,
|
||||
status,
|
||||
version,
|
||||
str(tool.priority)
|
||||
)
|
||||
|
||||
self.console.print(table)
|
||||
|
||||
if self.detected_tools:
|
||||
default = self.detected_tools[0]
|
||||
self.console.print(f"\n[green]Default tool: {default.display_name}[/green]")
|
||||
else:
|
||||
self.console.print("\n[yellow]No AI coding tools detected.[/yellow]")
|
||||
self.console.print("[dim]Install Claude Code, OpenAI CLI, or other tools to enable AI features.[/dim]")
|
||||
|
||||
def get_tool_command(self, tool: AITool, prompt: str) -> List[str]:
|
||||
"""Get the command to execute for a tool with a prompt."""
|
||||
if tool.name == "claude-code":
|
||||
return ["claude", prompt]
|
||||
elif tool.name == "hanzo-dev":
|
||||
return ["hanzo", "dev", "--prompt", prompt]
|
||||
elif tool.name == "openai-codex":
|
||||
return ["openai", "api", "completions.create", "-m", "code-davinci-002", "-p", prompt]
|
||||
elif tool.name == "gemini-cli":
|
||||
return ["gemini", "generate", "--prompt", prompt]
|
||||
elif tool.name == "grok-cli":
|
||||
return ["grok", "complete", prompt]
|
||||
elif tool.name == "openhands":
|
||||
return ["openhands", "run", prompt]
|
||||
elif tool.name == "cursor":
|
||||
return ["cursor", "--prompt", prompt]
|
||||
elif tool.name == "aider":
|
||||
return ["aider", "--message", prompt]
|
||||
else:
|
||||
return [tool.command, prompt]
|
||||
|
||||
def execute_with_tool(self, tool: AITool, prompt: str) -> Tuple[bool, str]:
|
||||
"""Execute a prompt with a specific tool."""
|
||||
try:
|
||||
command = self.get_tool_command(tool, prompt)
|
||||
result = subprocess.run(
|
||||
command,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
return True, result.stdout
|
||||
else:
|
||||
return False, result.stderr or "Command failed"
|
||||
except subprocess.TimeoutExpired:
|
||||
return False, "Command timed out"
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
|
||||
def execute_with_fallback(self, prompt: str) -> Tuple[bool, str, AITool]:
|
||||
"""Execute with fallback through available tools."""
|
||||
if not self.detected_tools:
|
||||
self.detect_all()
|
||||
|
||||
for tool in self.detected_tools:
|
||||
self.console.print(f"[dim]Trying {tool.display_name}...[/dim]")
|
||||
success, output = self.execute_with_tool(tool, prompt)
|
||||
|
||||
if success:
|
||||
return True, output, tool
|
||||
else:
|
||||
self.console.print(f"[yellow]{tool.display_name} failed: {output}[/yellow]")
|
||||
|
||||
return False, "No available tools could handle the request", None
|
||||
Reference in New Issue
Block a user