Hanzo Python SDK

The official Python SDK for the Hanzo AI platform - a complete AI infrastructure solution with unified gateway for 100+ LLM providers, cost tracking, rate limiting, and enterprise-ready observability.

🔥 NEW: Local AI Orchestration with 90% Cost Reduction

Hanzo Dev now supports local AI models as orchestrators, enabling you to use free/cheap local models to manage expensive API calls:

# Use local Llama 3.2 to orchestrate Claude, GPT-4, and Gemini
hanzo dev --orchestrator local:llama-3.2-3b --use-hanzo-net

# Your local AI decides when to use expensive APIs
# Result: 90% cost reduction while maintaining full capability

🚀 Features

• 100+ LLM Providers: OpenAI, Anthropic, Google, xAI (Grok), AWS Bedrock, Azure, Cohere, and more
• Smart Shell Tools: Automatic Zsh preference with fallback to Bash for enhanced command execution
• Local AI Orchestration: Use local models (Llama, Qwen, Mistral) to manage API usage
• Cost Optimization: 90% reduction through intelligent routing (local for simple, API for complex)
• Unified Interface: OpenAI-compatible API for all providers
• Enterprise Ready: Cost tracking, rate limiting, team management, and observability
• Type Safety: Full type hints and runtime validation with Pydantic
• Async Support: Both sync and async clients included
• 100% Test Coverage: Comprehensive test suite with 3,141 tests

📦 Installation

pip install hanzoai

For LiteLLM integration:

pip install hanzoai[litellm]

🎯 Quick Start

Basic Usage

from hanzoai import Hanzo

# Initialize the client
client = Hanzo(api_key="your-api-key")  # or set HANZO_API_KEY env var

# Make a chat completion request (OpenAI compatible)
response = client.chat.completions.create(
    model="gpt-4",  # or any supported model
    messages=[
        {"role": "user", "content": "Hello, how are you?"}
    ]
)

print(response.choices[0].message.content)

Using xAI Grok

from hanzoai import Hanzo

# Initialize client (requires XAI_API_KEY environment variable)
client = Hanzo(api_key="your-hanzo-api-key")

# Use Grok-4 for real-time knowledge and analysis
response = client.chat.completions.create(
    model="grok-4",  # or use aliases: "grok", "xai-grok"
    messages=[
        {"role": "user", "content": "What happened in tech news today?"}
    ]
)

print(response.choices[0].message.content)

# Grok with streaming for real-time responses
stream = client.chat.completions.create(
    model="grok-4",
    messages=[{"role": "user", "content": "Explain the latest AI developments"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")

Async Usage

import asyncio
from hanzoai import AsyncHanzo

async def main():
    client = AsyncHanzo(api_key="your-api-key")
    
    response = await client.chat.completions.create(
        model="claude-3-5-sonnet",
        messages=[
            {"role": "user", "content": "Explain quantum computing"}
        ]
    )
    
    print(response.choices[0].message.content)

asyncio.run(main())

Streaming Responses

from hanzoai import Hanzo

client = Hanzo(api_key="your-api-key")

# Stream chat completions
stream = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Count to 10"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")

🐚 Shell Tools (Zsh Default)

The SDK now includes smart shell selection that prefers Zsh when available:

Shell Tool (Smart Default)

from hanzo_mcp.tools.shell import shell_tool

# Automatically uses zsh if available, otherwise bash
result = await shell_tool.run(ctx, "echo $SHELL")

Explicit Shell Selection

from hanzo_mcp.tools.shell import zsh_tool, bash_tool

# Use Zsh explicitly (with enhanced features)
result = await zsh_tool.run(ctx, "echo ${(L)HOST}")  # Zsh parameter expansion

# Use Bash explicitly
result = await bash_tool.run(ctx, "echo $BASH_VERSION")

Shell Features

Zsh Advantages:

• Extended globbing: **/*.py recursively finds Python files
• Better completion and history
• Array and associative array support
• Rich plugin ecosystem (oh-my-zsh, etc.)
• Parameter expansion modifiers

Smart Shell Selection:

1. Checks for Zsh installation and .zshrc
2. Falls back to user's $SHELL preference
3. Defaults to Bash if nothing else available

CLI Usage

# Smart shell (prefers zsh)
hanzo-mcp shell "ls -la"

# Explicit zsh
hanzo-mcp zsh "echo $ZSH_VERSION"

# Explicit bash  
hanzo-mcp bash "echo $BASH_VERSION"

🤖 Using xAI Grok

Direct API Usage

from hanzoai import Hanzo
import os

# Set your xAI API key
os.environ["XAI_API_KEY"] = "your-xai-api-key"

# Initialize Hanzo client
client = Hanzo(api_key="your-hanzo-api-key")

# Use Grok for real-time knowledge
response = client.chat.completions.create(
    model="grok-4",  # Aliases: "grok", "xai-grok"
    messages=[
        {"role": "system", "content": "You have access to real-time information."},
        {"role": "user", "content": "What are the latest developments in AI today?"}
    ],
    temperature=0.7,
    max_tokens=2000
)

print(response.choices[0].message.content)

CLI Tool Usage

# Install hanzo-mcp for CLI tools
pip install hanzo-mcp

# Set your xAI API key
export XAI_API_KEY="your-xai-api-key"

# Use Grok CLI directly
grok "What's happening in the stock market right now?"

# Use Grok through hanzo dev
hanzo dev --orchestrator grok-4

# Batch operations with Grok
hanzo dev
> batch:10 agent:grok analyze real-time trends in *.py
> consensus:3 agent:grok,claude,gpt-4 review latest news

MCP Tool Integration

from hanzo_mcp.tools.agent.cli_tools import GrokCLITool
from mcp.server.fastmcp import Context

# Create Grok CLI tool
grok_tool = GrokCLITool()

# Use in your MCP server
async def analyze_realtime(ctx: Context, prompt: str):
    result = await grok_tool.call(
        ctx,
        prompt=prompt,
        model="grok-4",
        timeout=300
    )
    return result

Batch Processing with Grok

from hanzo.batch_orchestrator import BatchConfig, BatchOrchestrator
import asyncio

async def batch_grok_analysis():
    # Configure batch with Grok
    config = BatchConfig.from_command(
        "batch:20 agent:grok analyze market trends in data/*.csv"
    )
    
    orchestrator = BatchOrchestrator()
    results = await orchestrator.execute_batch(
        "batch:20 agent:grok real-time analysis"
    )
    
    return results

# Run batch analysis
results = asyncio.run(batch_grok_analysis())

Consensus with Grok and Other Models

# Use Grok in consensus with other models for balanced analysis
response = client.chat.completions.create(
    model="grok-4",
    messages=[{"role": "user", "content": "Analyze this code"}]
)

# Or use batch consensus
from hanzo.batch_orchestrator import BatchOrchestrator

orchestrator = BatchOrchestrator()
consensus_result = await orchestrator.execute_batch(
    "consensus:3 agent:grok,claude,gemini analyze security vulnerabilities"
)

Grok Features and Capabilities

• Real-time Knowledge: Access to current events and latest information
• Web Search Integration: Can search and analyze web content
• Multi-turn Conversations: Maintains context across exchanges
• Code Analysis: Strong programming and debugging capabilities
• Humor Mode: Can engage with wit and humor when appropriate
• Long Context: Handles up to 128K tokens context window

Environment Setup for Grok

# Required environment variables
export XAI_API_KEY="xai-..."  # Your xAI API key
export HANZO_API_KEY="sk-..."  # Your Hanzo API key (optional)

# Test Grok is working
python -c "
from hanzoai import Hanzo
import os
client = Hanzo()
response = client.chat.completions.create(
    model='grok-4',
    messages=[{'role': 'user', 'content': 'Hello Grok!'}]
)
print(response.choices[0].message.content)
"

🛠️ Advanced Features

Team & Organization Management

# Create a new team
team = client.team.create(
    team_alias="engineering",
    models=["gpt-4", "claude-3-5-sonnet"],
    max_budget=1000.0,
    rpm_limit=100
)

# Add members to team
client.team.add_member(
    team_id=team.team_id,
    member=[{"user_email": "[email protected]", "role": "user"}]
)

# Track spending
spend_report = client.spend.list_logs()

Model Management

# List available models
models = client.models.list()

# Get model info
model_info = client.model.info.get(model="gpt-4")

# Create custom model configuration
client.models.create(
    model_name="my-custom-gpt4",
    hanzo_params={
        "model": "gpt-4",
        "temperature": 0.7,
        "max_tokens": 2000,
        "api_base": "https://api.openai.com/v1"
    }
)

File Operations

# Upload a file
file = client.files.create(
    file=open("data.jsonl", "rb"),
    purpose="fine-tune"
)

# List files
files = client.files.list()

# Get file content
content = client.files.content.get(file_id=file.id)

Fine-tuning

# Create a fine-tuning job
job = client.fine_tuning.jobs.create(
    model="gpt-3.5-turbo",
    training_file=file.id,
    hyperparameters={
        "n_epochs": 3,
        "batch_size": 1,
        "learning_rate_multiplier": 1.0
    }
)

# Monitor job status
status = client.fine_tuning.jobs.retrieve(job_id=job.id)
print(f"Status: {status.status}")

# List all jobs
jobs = client.fine_tuning.jobs.list()

Embeddings

# Generate embeddings
response = client.embeddings.create(
    model="text-embedding-3-small",
    input=["Hello world", "How are you?"]
)

for embedding in response.data:
    print(f"Embedding dimension: {len(embedding.embedding)}")

Image Generation

# Generate images
response = client.images.generate(
    model="dall-e-3",
    prompt="A futuristic city at sunset",
    n=1,
    size="1024x1024"
)

print(response.data[0].url)

Audio Transcription

# Transcribe audio
transcript = client.audio.transcriptions.create(
    model="whisper-1",
    file=open("audio.mp3", "rb")
)

print(transcript.text)

🔧 Configuration

Environment Variables

export HANZO_API_KEY="your-api-key"
export HANZO_BASE_URL="https://api.hanzo.ai"  # optional
export HANZO_LOG="info"  # Enable logging (debug/info/warning/error)

Custom HTTP Client

import httpx
from hanzoai import Hanzo, DefaultHttpxClient

client = Hanzo(
    api_key="your-api-key",
    http_client=DefaultHttpxClient(
        proxy="http://proxy.example.com:8080",
        timeout=30.0,
        limits=httpx.Limits(max_connections=100)
    )
)

Retry Configuration

from hanzoai import Hanzo

client = Hanzo(
    api_key="your-api-key",
    max_retries=3,  # Default is 2
    timeout=60.0     # Default is 60 seconds
)

# Per-request configuration
response = client.with_options(
    max_retries=5,
    timeout=120.0
).chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello"}]
)

📊 Error Handling

from hanzoai import Hanzo
import hanzoai

client = Hanzo(api_key="your-api-key")

try:
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Hello"}]
    )
except hanzoai.APIConnectionError as e:
    print(f"Connection error: {e}")
except hanzoai.RateLimitError as e:
    print(f"Rate limit exceeded: {e}")
    print(f"Retry after: {e.response.headers.get('retry-after')}")
except hanzoai.APIStatusError as e:
    print(f"API error: {e.status_code}")
    print(f"Response: {e.response}")

Error Types

Error Type	Description
APIConnectionError	Network connectivity issues
APITimeoutError	Request timeout
RateLimitError	Rate limit exceeded (429)
AuthenticationError	Invalid API key (401)
PermissionDeniedError	Insufficient permissions (403)
NotFoundError	Resource not found (404)
UnprocessableEntityError	Invalid request (422)
InternalServerError	Server error (500+)

🧪 Development

Setup Development Environment

# Clone the repository
git clone https://github.com/hanzoai/python-sdk.git
cd python-sdk

# Install with uv (recommended)
uv pip install -e .

# Or with pip
pip install -e .

# Install dev dependencies
uv pip install -r requirements-dev.lock

Running Tests

# Run all tests
./scripts/test

# Or with pytest directly
uv run pytest tests/

# Run with coverage
uv run pytest --cov=hanzoai tests/

# Run specific test file
uv run pytest tests/api_resources/test_chat.py

Code Quality

# Run lints
./scripts/lint

# Format code
uv run ruff format pkg/

# Type checking
uv run mypy pkg/

📚 Documentation

• API Reference: Full API documentation at docs.hanzo.ai
• SDK Reference: Detailed SDK reference in api.md
• Examples: See the examples/ directory
• Contributing: Read CONTRIBUTING.md

🤝 Support

• Issues: GitHub Issues
• Discussions: GitHub Discussions
• Discord: Join our Discord
• Email: [email protected]

📄 License

This project is licensed under the BSD-3-Clause License. See LICENSE file for details.

🏗️ Project Status

• ✅ 3,141 tests - 100% passing (all integration tests now use mocks)
• ✅ Type Safe - Full type hints with Pydantic validation
• ✅ Production Ready - Used in enterprise deployments
• ✅ Active Development - Regular updates and improvements
• ✅ CI/CD - Automated testing and deployment pipeline
• ✅ No External Dependencies - All tests run without external services

🌟 Contributors

Thanks to all our contributors! See CONTRIBUTORS.md for the full list.

---

Built with ❤️ by Hanzo AI