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

feat(cli): Add Hanzo Node as highest priority for local private AI

Browse Source 
- Hanzo Node (hanzod) now highest priority when running on localhost:8000
- Provides completely private, local AI - data never leaves your machine
- Detects running Hanzo Router on localhost:4000 as LLM proxy
- Priority chain: Hanzo Node → Router → Claude Code → other tools
- Shows model count when Hanzo Node is running
- Special messaging for privacy when using local AI
- Version 0.3.27 with enhanced local AI support
This commit is contained in:
zeekay
2025-09-06 06:49:36 -05:00
parent ead6d01ff5
commit f267b17c16
5 changed files with 993 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
WORKFLOW_GUIDE.md
+622
View File
@@ -0,0 +1,622 @@
# Hanzo AI Peak Workflow Guide
## Overview
This guide documents the complete Hanzo AI workflow for architecture, planning, task generation, and development using Claude, MCP tools, and the Hanzo SDK.
## Prerequisites
### Installation
```bash
# Install Hanzo SDK and all components
pip install hanzo[all]
# Or install specific components
pip install hanzo hanzo-mcp hanzo-agents hanzo-repl
# Verify installation
hanzo --version
hanzo-mcp --help
```
### Environment Setup
```bash
# Required API keys
export ANTHROPIC_API_KEY="sk-ant-..."
export OPENAI_API_KEY="sk-..." # Optional
export HANZO_API_KEY="..." # Optional for Hanzo Router
# Optional configuration
export HANZO_DEFAULT_MODEL="claude-3-5-sonnet-20241022"
export HANZO_USE_LOCAL="false"
```
## Phase 1: Architecture & Planning
### 1.1 Independent Architecture Proposals
```bash
# Using Hanzo REPL with Claude agent
hanzo repl start --model claude-3-opus-20240229
# In REPL, request architecture proposal
agent:claude "Analyze the codebase and propose architecture for [FEATURE]"
# Or use multiple agents for consensus
swarm:3 "Independently analyze and propose architecture for [FEATURE], then build consensus"
```
### 1.2 Architecture Refinement
```python
# Using Python SDK for architecture refinement
from hanzoai import Hanzo
import json
client = Hanzo()
# Get initial proposals
proposals = []
for agent in ["claude-3-opus", "claude-3-5-sonnet", "gpt-4"]:
response = client.chat.completions.create(
model=agent,
messages=[
{"role": "system", "content": "You are a senior architect"},
{"role": "user", "content": f"Propose architecture for: {feature_description}"}
]
)
proposals.append(response.choices[0].message.content)
# Refine with GPT-5 Pro (or Claude Opus)
refined = client.chat.completions.create(
model="claude-3-opus-20240229",
messages=[
{"role": "system", "content": "You are a principal architect reviewing proposals"},
{"role": "user", "content": f"Review and improve these proposals:\n{json.dumps(proposals, indent=2)}"}
]
)
# Save architecture document
with open("ARCHITECTURE.md", "w") as f:
f.write(refined.choices[0].message.content)
```
### 1.3 Save Architecture with MCP
```bash
# Using hanzo-mcp to save architecture
hanzo mcp write --file ARCHITECTURE.md --content "$ARCHITECTURE_CONTENT"
# Or in Python
from hanzo_mcp import MCPClient
mcp = MCPClient()
mcp.write("ARCHITECTURE.md", architecture_content)
```
## Phase 2: Task Generation
### 2.1 Generate Linear Tasks
```python
# Read architecture and generate tasks
from hanzoai import Hanzo
from hanzo_mcp import MCPClient
client = Hanzo()
mcp = MCPClient()
# Read architecture
architecture = mcp.read("ARCHITECTURE.md")
# Generate tasks
tasks_response = client.chat.completions.create(
model="claude-3-5-sonnet-20241022",
messages=[
{"role": "system", "content": "You are a technical project manager"},
{"role": "user", "content": f"""
Based on this architecture:
{architecture}
Generate small, focused Linear tasks that a junior engineer can complete in under a day.
Format as JSON with: title, description, priority, dependencies, acceptance_criteria
"""}
]
)
tasks = json.loads(tasks_response.choices[0].message.content)
```
### 2.2 Refine Tasks with GPT-5 Pro
```python
# Refine task list
refined_tasks = client.chat.completions.create(
model="claude-3-opus-20240229", # Or GPT-5 when available
messages=[
{"role": "system", "content": "You are an engineering manager"},
{"role": "user", "content": f"""
Review and improve these tasks:
{json.dumps(tasks, indent=2)}
Ensure:
- Clear scope and boundaries
- Proper dependency ordering
- Testable acceptance criteria
- Realistic time estimates
"""}
]
)
```
### 2.3 Create Linear Issues (with MCP Linear integration)
```python
# If Linear MCP is available
from hanzo_mcp.tools import linear
# Create project
project = linear.create_project(
name="Feature Implementation",
description=architecture_summary
)
# Create epic
epic = linear.create_issue(
title="Main Feature Epic",
description=architecture,
project_id=project.id,
type="Epic"
)
# Create tasks
for task in refined_tasks:
issue = linear.create_issue(
title=task["title"],
description=task["description"],
project_id=project.id,
parent_id=epic.id,
priority=task["priority"],
labels=task.get("labels", [])
)
# Set dependencies
for dep_title in task.get("dependencies", []):
dep_issue = linear.find_issue(title=dep_title)
if dep_issue:
linear.add_dependency(issue.id, dep_issue.id)
```
## Phase 3: Development Workflow
### 3.1 Create Worktree per Task
```bash
#!/bin/bash
# Script: create_task_worktree.sh
TASK_ID=$1
TASK_TITLE=$2
# Create worktree
git worktree add -b "task-$TASK_ID" "./worktrees/$TASK_ID" main
cd "./worktrees/$TASK_ID"
# Save task context
echo "Task: $TASK_TITLE" > .task_context
echo "Linear ID: $TASK_ID" >> .task_context
```
### 3.2 One-Shot Task Implementation
```python
# one_shot_task.py
import sys
from hanzoai import Hanzo
from hanzo_mcp import MCPClient
def implement_task(task_id: str):
client = Hanzo()
mcp = MCPClient()
# Get task details (from Linear or local file)
task = load_task(task_id)
architecture = mcp.read("ARCHITECTURE.md")
# One-shot implementation
implementation = client.chat.completions.create(
model="claude-3-5-sonnet-20241022",
messages=[
{"role": "system", "content": f"""
You are implementing a specific task.
Architecture: {architecture}
Use MCP tools to read, write, and test code.
"""},
{"role": "user", "content": f"""
Implement this task:
{json.dumps(task, indent=2)}
Requirements:
1. Follow the architecture exactly
2. Write comprehensive tests
3. Add proper documentation
4. Ensure all tests pass
"""}
],
tools=mcp.get_all_tools(),
tool_choice="auto"
)
return implementation
if __name__ == "__main__":
task_id = sys.argv[1]
result = implement_task(task_id)
print(f"Task {task_id} implemented")
```
### 3.3 Code Review with Claude Code
```python
# review_task.py
def review_implementation(task_id: str):
client = Hanzo()
mcp = MCPClient()
# Get changed files
changes = mcp.run_command("git diff main")
# Review with Claude Code
review = client.chat.completions.create(
model="claude-3-opus-20240229",
messages=[
{"role": "system", "content": "You are a senior code reviewer"},
{"role": "user", "content": f"""
Review these changes:
{changes}
Check for:
- Bugs and edge cases
- Performance issues
- Security vulnerabilities
- Code style and best practices
- Test coverage
"""}
]
)
return review.choices[0].message.content
```
### 3.4 Iterative Refinement
```python
# refine_implementation.py
def refine_with_feedback(task_id: str, feedback: str):
client = Hanzo()
mcp = MCPClient()
refinement = client.chat.completions.create(
model="claude-3-5-sonnet-20241022",
messages=[
{"role": "system", "content": "You are addressing code review feedback"},
{"role": "user", "content": f"""
Address this feedback:
{feedback}
Make necessary changes using MCP tools.
"""}
],
tools=mcp.get_all_tools(),
tool_choice="auto"
)
return refinement
```
## Phase 4: Pull Request Workflow
### 4.1 Create Pull Request
```bash
# Push changes and create PR
git push -u origin "task-$TASK_ID"
# Create PR with gh CLI
gh pr create \
--title "Task $TASK_ID: $TASK_TITLE" \
--body "$(cat .task_context)" \
--base main
```
### 4.2 Automated PR Review
```python
# pr_review_bot.py
def review_pr(pr_number: int):
reviews = []
# Get PR diff
diff = mcp.run_command(f"gh pr diff {pr_number}")
# Review with multiple agents
for model in ["claude-3-opus", "claude-3-5-sonnet"]:
review = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": "Review this PR for issues"},
{"role": "user", "content": diff}
]
)
reviews.append(review.choices[0].message.content)
# Post reviews as comments
for review in reviews:
mcp.run_command(f'gh pr comment {pr_number} --body "{review}"')
return reviews
```
## Complete Automation Script
### Full Workflow Automation
```python
#!/usr/bin/env python3
# hanzo_workflow.py
import json
import os
import sys
from pathlib import Path
from typing import List, Dict, Any
from hanzoai import Hanzo
from hanzo_mcp import MCPClient
from hanzo_agents import SwarmOrchestrator
class HanzoWorkflow:
def __init__(self):
self.client = Hanzo()
self.mcp = MCPClient()
self.swarm = SwarmOrchestrator()
def run_architecture_phase(self, feature: str) -> str:
"""Phase 1: Architecture & Planning"""
print("🏗️ Phase 1: Architecture & Planning")
# Get proposals from multiple agents
proposals = self.swarm.run(
agents=["claude-3-opus", "claude-3-5-sonnet", "gpt-4"],
task=f"Propose architecture for: {feature}",
consensus=True
)
# Refine with top model
refined = self.client.chat.completions.create(
model="claude-3-opus-20240229",
messages=[
{"role": "system", "content": "You are a principal architect"},
{"role": "user", "content": f"Refine: {proposals}"}
]
)
# Save architecture
architecture = refined.choices[0].message.content
self.mcp.write("ARCHITECTURE.md", architecture)
print("✅ Architecture saved to ARCHITECTURE.md")
return architecture
def run_task_generation(self, architecture: str) -> List[Dict]:
"""Phase 2: Task Generation"""
print("📋 Phase 2: Task Generation")
# Generate tasks
tasks_response = self.client.chat.completions.create(
model="claude-3-5-sonnet-20241022",
messages=[
{"role": "system", "content": "Generate Linear tasks"},
{"role": "user", "content": f"From architecture: {architecture}"}
],
response_format={"type": "json_object"}
)
tasks = json.loads(tasks_response.choices[0].message.content)
# Save tasks
self.mcp.write("TASKS.json", json.dumps(tasks, indent=2))
print(f"✅ Generated {len(tasks)} tasks")
return tasks
def implement_task(self, task: Dict) -> bool:
"""Phase 3: Task Implementation"""
task_id = task["id"]
print(f"💻 Implementing task {task_id}: {task['title']}")
# Create worktree
worktree_path = f"./worktrees/{task_id}"
self.mcp.run_command(f"git worktree add -b task-{task_id} {worktree_path} main")
# Implement with Claude
os.chdir(worktree_path)
implementation = self.client.chat.completions.create(
model="claude-3-5-sonnet-20241022",
messages=[
{"role": "system", "content": "Implement this task using MCP tools"},
{"role": "user", "content": json.dumps(task)}
],
tools=self.mcp.get_all_tools(),
tool_choice="auto"
)
# Run tests
test_result = self.mcp.run_command("make test")
if "PASSED" in test_result:
print(f"✅ Task {task_id} implemented successfully")
return True
else:
print(f"❌ Task {task_id} tests failed")
return False
def create_pr(self, task: Dict) -> str:
"""Phase 4: Create Pull Request"""
task_id = task["id"]
print(f"🔀 Creating PR for task {task_id}")
# Push branch
self.mcp.run_command(f"git push -u origin task-{task_id}")
# Create PR
pr_output = self.mcp.run_command(f"""
gh pr create \
--title "Task {task_id}: {task['title']}" \
--body "{task['description']}" \
--base main
""")
pr_number = pr_output.split("/")[-1].strip()
print(f"✅ Created PR #{pr_number}")
return pr_number
def run_full_workflow(self, feature: str):
"""Run complete workflow"""
print("🚀 Starting Hanzo AI Peak Workflow")
print("="*50)
# Phase 1: Architecture
architecture = self.run_architecture_phase(feature)
# Phase 2: Task Generation
tasks = self.run_task_generation(architecture)
# Phase 3 & 4: Implementation and PRs
for task in tasks:
if self.implement_task(task):
self.create_pr(task)
# Return to main directory
os.chdir(Path(__file__).parent)
print("="*50)
print("✨ Workflow complete!")
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: hanzo_workflow.py 'Feature description'")
sys.exit(1)
feature = sys.argv[1]
workflow = HanzoWorkflow()
workflow.run_full_workflow(feature)
```
## Using Hanzo CLI and REPL
### Interactive Chat with Agent Selection
```bash
# Start REPL
hanzo repl start
# In REPL, use different agent syntaxes:
> agent:claude "Analyze this code for security issues"
> agent:gpt4 "Generate unit tests"
> swarm:3 "Brainstorm solutions for this problem"
# Direct model selection
> @claude-3-opus How would you architect this system?
> @gpt-4 What are the performance implications?
```
### CLI Quick Commands
```bash
# Quick questions
hanzo ask "How do I implement OAuth2 in FastAPI?"
# Chat with specific model
hanzo chat --model claude-3-5-sonnet-20241022
# Use local models
hanzo chat --local --model llama-3.2-3b
# With MCP tools
hanzo chat --tools filesystem,search,git
```
## Advanced Features
### Voice Mode
```bash
# Enable voice in REPL
hanzo repl start --voice
# Or set environment
export HANZO_ENABLE_VOICE=true
hanzo repl start
```
### TUI Interface
```bash
# Beautiful terminal UI
hanzo repl start --tui
```
### IPython Integration
```bash
# IPython magic commands
hanzo repl start --ipython
# In IPython:
%mcp_read file.py
%mcp_search "pattern"
%agent claude "Analyze this"
```
## Environment Variables
```bash
# Core configuration
export ANTHROPIC_API_KEY="sk-ant-..."
export OPENAI_API_KEY="sk-..."
export HANZO_API_KEY="..."
# Default settings
export HANZO_DEFAULT_MODEL="claude-3-5-sonnet-20241022"
export HANZO_USE_LOCAL="false"
export HANZO_ENABLE_VOICE="false"
# Router configuration (if using Hanzo Router)
export HANZO_ROUTER_URL="http://localhost:4000"
export HANZO_USE_ROUTER="true"
```
## Troubleshooting
### Common Issues
1. **Import errors**: Ensure all packages are installed
```bash
pip install hanzo[all]
```
2. **API key issues**: Check environment variables
```bash
echo $ANTHROPIC_API_KEY
echo $OPENAI_API_KEY
```
3. **MCP tools not available**: Install hanzo-mcp
```bash
pip install hanzo-mcp
hanzo-mcp install-desktop # For Claude Desktop
```
4. **REPL not starting**: Check dependencies
```bash
pip install hanzo-repl ipython textual
```
## Summary
This workflow enables:
1. ✅ Architecture generation with multiple AI models
2. ✅ Task breakdown and Linear integration
3. ✅ Automated implementation with one-shot prompts
4. ✅ Code review and iterative refinement
5. ✅ PR creation and automated review
6. ✅ Full CLI and REPL support with agent:claude syntax
7. ✅ Swarm orchestration for consensus building
8. ✅ MCP tool integration throughout
The system is fully functional and can be used immediately with proper API keys configured.
pkg/hanzo/pyproject.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "hanzo"
version = "0.3.26"
version = "0.3.27"
description = "Hanzo AI - Complete AI Infrastructure Platform with CLI, Router, MCP, and Agent Runtime"
authors = [
{name = "Hanzo AI", email = "[email protected]"},
pkg/hanzo/src/hanzo/__init__.py
+1 -1
View File
@@ -1,6 +1,6 @@
"""Hanzo - Complete AI Infrastructure Platform with CLI, Router, MCP, and Agent Runtime."""
__version__ = "0.3.26"
__version__ = "0.3.27"
__all__ = ["main", "cli", "__version__"]
from .cli import cli, main
pkg/hanzo/src/hanzo/tools/detector.py
+127 -31
View File
@@ -3,6 +3,7 @@
import os
import shutil
import subprocess
import httpx
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
@@ -33,12 +34,33 @@ class ToolDetector:
# Define available tools with priority order
TOOLS = [
# Hanzo Local Node - highest priority for privacy and local control
AITool(
name="hanzod",
command="hanzo node",
display_name="Hanzo Node (Local Private AI)",
provider="hanzo-local",
priority=0, # Highest priority - local and private
check_command=None, # Check via API endpoint
api_endpoint="http://localhost:8000/health",
env_var=None
),
AITool(
name="hanzo-router",
command="hanzo router",
display_name="Hanzo Router (LLM Proxy)",
provider="hanzo-router",
priority=1,
check_command=None,
api_endpoint="http://localhost:4000/health",
env_var=None
),
AITool(
name="claude-code",
command="claude",
display_name="Claude Code",
provider="anthropic",
priority=1,
priority=2,
check_command="claude --version",
env_var="ANTHROPIC_API_KEY"
),
@@ -47,7 +69,7 @@ class ToolDetector:
command="hanzo dev",
display_name="Hanzo Dev (Native)",
provider="hanzo",
priority=2,
priority=3,
check_command="hanzo --version",
env_var="HANZO_API_KEY"
),
@@ -56,7 +78,7 @@ class ToolDetector:
command="openai",
display_name="OpenAI Codex",
provider="openai",
priority=3,
priority=4,
check_command="openai --version",
env_var="OPENAI_API_KEY"
),
@@ -65,7 +87,7 @@ class ToolDetector:
command="gemini",
display_name="Gemini CLI",
provider="google",
priority=4,
priority=5,
check_command="gemini --version",
env_var="GEMINI_API_KEY"
),
@@ -74,7 +96,7 @@ class ToolDetector:
command="grok",
display_name="Grok CLI",
provider="xai",
priority=5,
priority=6,
check_command="grok --version",
env_var="GROK_API_KEY"
),
@@ -83,7 +105,7 @@ class ToolDetector:
command="openhands",
display_name="OpenHands CLI",
provider="openhands",
priority=6,
priority=7,
check_command="openhands --version",
env_var=None
),
@@ -92,7 +114,7 @@ class ToolDetector:
command="cursor",
display_name="Cursor AI",
provider="cursor",
priority=7,
priority=8,
check_command="cursor --version",
env_var=None
),
@@ -101,7 +123,7 @@ class ToolDetector:
command="codeium",
display_name="Codeium",
provider="codeium",
priority=8,
priority=9,
check_command="codeium --version",
env_var="CODEIUM_API_KEY"
),
@@ -110,7 +132,7 @@ class ToolDetector:
command="aider",
display_name="Aider",
provider="aider",
priority=9,
priority=10,
check_command="aider --version",
env_var=None
),
@@ -119,7 +141,7 @@ class ToolDetector:
command="continue",
display_name="Continue Dev",
provider="continue",
priority=10,
priority=11,
check_command="continue --version",
env_var=None
)
@@ -143,26 +165,51 @@ class ToolDetector:
def detect_tool(self, tool: AITool) -> bool:
"""Detect if a specific tool is available."""
# Check API endpoint first (for services like hanzod)
if tool.api_endpoint:
try:
response = httpx.get(tool.api_endpoint, timeout=1.0)
if response.status_code == 200:
tool.detected = True
tool.version = "Running"
# Special handling for Hanzo services
if tool.name == "hanzod":
# Check if models are loaded
try:
models_response = httpx.get("http://localhost:8000/models", timeout=1.0)
if models_response.status_code == 200:
models = models_response.json()
if models:
tool.version = f"Running ({len(models)} models)"
except:
pass
return True
except:
pass
# 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
if tool.command:
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):
@@ -229,13 +276,25 @@ class ToolDetector:
if self.detected_tools:
default = self.detected_tools[0]
self.console.print(f"\n[green]Default tool: {default.display_name}[/green]")
# Special message for Hanzo Node
if default.name == "hanzod":
self.console.print("[cyan]🔒 Using local private AI - your data stays on your machine[/cyan]")
self.console.print("[dim]Manage models with: hanzo node models[/dim]")
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]")
self.console.print("[dim]Start Hanzo Node for local AI: hanzo node start[/dim]")
self.console.print("[dim]Or install Claude Code, OpenAI CLI, etc.[/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":
if tool.name == "hanzod":
# Use the local Hanzo node API
return ["hanzo", "ask", "--local", prompt]
elif tool.name == "hanzo-router":
# Use the router proxy
return ["hanzo", "ask", "--router", prompt]
elif tool.name == "claude-code":
return ["claude", prompt]
elif tool.name == "hanzo-dev":
return ["hanzo", "dev", "--prompt", prompt]
@@ -257,6 +316,43 @@ class ToolDetector:
def execute_with_tool(self, tool: AITool, prompt: str) -> Tuple[bool, str]:
"""Execute a prompt with a specific tool."""
try:
# Special handling for Hanzo services
if tool.name == "hanzod":
# Use the local API directly
try:
response = httpx.post(
"http://localhost:8000/chat/completions",
json={
"messages": [{"role": "user", "content": prompt}],
"stream": False
},
timeout=30.0
)
if response.status_code == 200:
result = response.json()
return True, result.get("choices", [{}])[0].get("message", {}).get("content", "")
except Exception as e:
return False, f"Hanzo Node error: {e}"
elif tool.name == "hanzo-router":
# Use the router API
try:
response = httpx.post(
"http://localhost:4000/chat/completions",
json={
"messages": [{"role": "user", "content": prompt}],
"model": "gpt-3.5-turbo", # Router will route to best available
"stream": False
},
timeout=30.0
)
if response.status_code == 200:
result = response.json()
return True, result.get("choices", [{}])[0].get("message", {}).get("content", "")
except Exception as e:
return False, f"Router error: {e}"
# Default command execution
command = self.get_tool_command(tool, prompt)
result = subprocess.run(
command,
test_workflow.py
Executable
+242
View File
@@ -0,0 +1,242 @@
#!/usr/bin/env python3
"""Test script to validate the Hanzo workflow is properly configured."""
import os
import sys
import json
from pathlib import Path
def check_environment():
"""Check required environment variables."""
print("🔍 Checking environment variables...")
required = {
"ANTHROPIC_API_KEY": "Anthropic API key for Claude models",
"OPENAI_API_KEY": "OpenAI API key (optional but recommended)",
}
optional = {
"HANZO_API_KEY": "Hanzo Router API key",
"HANZO_DEFAULT_MODEL": "Default model selection",
"HANZO_ROUTER_URL": "Hanzo Router URL",
}
missing = []
for key, desc in required.items():
if os.environ.get(key):
print(f" ✅ {key}: Set")
else:
print(f" ❌ {key}: Missing - {desc}")
missing.append(key)
print("\nOptional variables:")
for key, desc in optional.items():
value = os.environ.get(key)
if value:
print(f" ✅ {key}: {value[:20]}...")
else:
print(f" ⚠️ {key}: Not set - {desc}")
return len(missing) == 0
def test_imports():
"""Test that all required packages can be imported."""
print("\n📦 Testing package imports...")
packages = [
("hanzoai", "Core SDK"),
("hanzo_mcp", "MCP tools"),
("hanzo_agents", "Agent framework"),
("hanzo_repl", "REPL interface"),
]
failed = []
for package, desc in packages:
try:
__import__(package)
print(f" ✅ {package}: {desc}")
except ImportError as e:
print(f" ❌ {package}: {desc} - {e}")
failed.append(package)
return len(failed) == 0
def test_basic_client():
"""Test basic Hanzo client functionality."""
print("\n🤖 Testing Hanzo client...")
try:
from hanzoai import Hanzo
client = Hanzo()
print(" ✅ Client initialized")
# Test model listing (doesn't require API call)
models = ["claude-3-opus-20240229", "claude-3-5-sonnet-20241022", "gpt-4"]
print(f" ✅ Available models: {', '.join(models[:3])}...")
return True
except Exception as e:
print(f" ❌ Client error: {e}")
return False
def test_mcp_tools():
"""Test MCP tool availability."""
print("\n🛠️ Testing MCP tools...")
try:
from hanzo_mcp import MCPClient
mcp = MCPClient()
tools = mcp.get_all_tools()
print(f" ✅ {len(tools)} MCP tools available")
# Check for essential tools
essential = ["read", "write", "search", "run_command"]
available = [t['name'] for t in tools]
for tool in essential:
if any(tool in t for t in available):
print(f" ✅ Tool '{tool}' available")
else:
print(f" ⚠️ Tool '{tool}' might be missing")
return True
except Exception as e:
print(f" ❌ MCP error: {e}")
return False
def test_agent_framework():
"""Test agent framework setup."""
print("\n🤝 Testing agent framework...")
try:
from hanzo_agents import Agent, SwarmOrchestrator
# Test agent creation
agent = Agent(name="test", model="claude-3-haiku-20240307")
print(" ✅ Agent created")
# Test swarm
swarm = SwarmOrchestrator()
print(" ✅ Swarm orchestrator ready")
return True
except Exception as e:
print(f" ❌ Agent framework error: {e}")
return False
def test_cli_commands():
"""Test CLI command availability."""
print("\n💻 Testing CLI commands...")
import subprocess
commands = [
("hanzo --version", "Hanzo CLI"),
("hanzo-mcp --help", "MCP CLI"),
]
failed = []
for cmd, desc in commands:
try:
result = subprocess.run(
cmd.split(),
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
print(f" ✅ {desc}: Available")
else:
print(f" ⚠️ {desc}: Non-zero exit code")
except Exception as e:
print(f" ❌ {desc}: {e}")
failed.append(cmd)
return len(failed) == 0
def test_simple_completion():
"""Test a simple completion if API key is available."""
print("\n🧪 Testing simple completion...")
if not os.environ.get("ANTHROPIC_API_KEY"):
print(" ⏭️ Skipping - No API key")
return True
try:
from hanzoai import Hanzo
client = Hanzo()
response = client.chat.completions.create(
model="claude-3-haiku-20240307", # Cheapest model
messages=[{"role": "user", "content": "Say 'test successful' only"}],
max_tokens=10
)
result = response.choices[0].message.content.lower()
if "test" in result or "successful" in result:
print(" ✅ Completion successful")
return True
else:
print(f" ⚠️ Unexpected response: {result}")
return True
except Exception as e:
print(f" ❌ Completion error: {e}")
return False
def main():
"""Run all tests."""
print("=" * 50)
print("🚀 Hanzo Workflow Test Suite")
print("=" * 50)
tests = [
("Environment", check_environment),
("Imports", test_imports),
("Client", test_basic_client),
("MCP Tools", test_mcp_tools),
("Agents", test_agent_framework),
("CLI", test_cli_commands),
("Completion", test_simple_completion),
]
results = {}
for name, test_func in tests:
try:
results[name] = test_func()
except Exception as e:
print(f"\n❌ Test '{name}' crashed: {e}")
results[name] = False
# Summary
print("\n" + "=" * 50)
print("📊 Test Summary")
print("=" * 50)
passed = sum(1 for r in results.values() if r)
total = len(results)
for name, result in results.items():
status = "✅ PASS" if result else "❌ FAIL"
print(f" {status}: {name}")
print(f"\n🎯 Result: {passed}/{total} tests passed")
if passed == total:
print("✨ All tests passed! The workflow is ready to use.")
print("\nNext steps:")
print(" 1. Start REPL: hanzo repl start")
print(" 2. Try agent:claude syntax in REPL")
print(" 3. Run the workflow script: python hanzo_workflow.py 'Your feature'")
else:
print("\n⚠️ Some tests failed. Please check:")
print(" 1. Install missing packages: pip install hanzo[all]")
print(" 2. Set required environment variables")
print(" 3. Check the WORKFLOW_GUIDE.md for details")
return 0 if passed == total else 1
if __name__ == "__main__":
sys.exit(main())