Files
python-sdk/pkg/hanzo-agent/docs/hanzo-backend-integration.md
zeekay 241ca6dc59 the front door tells the truth about the client behind it
README: PyPI serves 3.2.1 and this tree is 3.2.11, so the install line is the
git one. Between the two the default version left the operation ids — 3.2.1
carries get_v1_keys/get_v1_tools (182 api modules, 2172 models) and this carries
get_keys/get_tools — so `pip install hanzoai` above a `get_keys()` quickstart
hands the reader a client that cannot run it. Counts are measured off the pinned
document: 1814 paths, 2479 operations, 191 tags + default_api = 192 API classes
over 2460 models, 2659 importable modules.

Auth was wrong everywhere and it was the same root cause: the document declares
no securityScheme, so every operation got an empty auth_settings and
Configuration(access_token=…) is read by nothing. Serializing get_keys from such
a Configuration produces a request with no Authorization header at all — the
five flows have been calling anonymously and reading the refusal as a bad key.
examples/client.py now passes the bearer through ApiClient's header_name/value
pair, verified against a local server that echoes what it received, and a smoke
test pins both halves so the day cloud emits the scheme, the test fails and the
header moves back into Configuration.

The examples gate resolved class names and stopped there, which is how `money`
and `tools` once passed it while every call in them named a method that no
longer existed. It now walks the syntax tree for attribute accesses on anything
bound to a *Api and asks the class: ten names resolve, and a stale get_v1_tools
fails in the same second the import passes.

Deleted five example scripts that imported hanzoai.Hanzo, hanzoai.cluster,
hanzoai.agents and hanzoai.completion — none of which exist — or were dict
literals captioned "in practice this would be called through the MCP interface";
compileall passed all five, because syntax is not a claim. Deleted three docs
that were fabrication rather than drift and that nothing rendered: FEATURES.md,
QUICKSTART.md, GPT5_ORCHESTRATION.md.

pkg/hanzo-agent said `pip install hanzoai` and `from hanzoai import Agent,
Runner`; the client has no Agent. Its own `full` extra resolved to
hanzoai[web3,tee,marketplace,cli], extras the client does not declare, so it
installed the client and none of the extensions. pyproject said BSD-3-Clause
while LICENSE is the Apache 2.0 text. test_smoke imported tracker_api and no
/v1/tracker path is emitted any more, so the suite carried a failing test naming
a module the client does not have; it reads analytics_api now, off the client.
2026-08-14 14:31:08 -07:00

7.6 KiB

Hanzo Agent SDK - Backend Integration Guide

This guide explains how to configure the Hanzo Agent SDK to work with your Hanzo AI infrastructure instead of calling OpenAI directly.

Overview

The Hanzo Agent SDK can be configured to route all LLM requests through the Hanzo Router, which provides:

  • Unified Access: Connect to 100+ LLM providers through a single API
  • Cost Management: Track usage and costs across all models
  • Reliability: Automatic fallbacks and load balancing
  • Observability: Monitor all requests via the Cloud dashboard
  • Security: Keep API keys secure in your infrastructure

Quick Start

1. Install the SDK

pip install hanzo-agent
# or
uv pip install hanzo-agent

2. Configure Environment

# Local development
export HANZO_ROUTER_URL="http://localhost:4000/v1"
export HANZO_API_KEY="sk-1234"  # Get from Router dashboard

# Production
export HANZO_ROUTER_URL="https://router.your-domain.com/v1"
export HANZO_API_KEY="sk-production-key"

3. Basic Usage

from openai import AsyncOpenAI
from agents import Agent, Runner, RunConfig, ModelProvider, Model
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel

class HanzoModelProvider(ModelProvider):
    def __init__(self, base_url: str, api_key: str):
        self.client = AsyncOpenAI(base_url=base_url, api_key=api_key)
    
    def get_model(self, model_name: str | None) -> Model:
        return OpenAIChatCompletionsModel(
            model=model_name or "gpt-3.5-turbo",
            openai_client=self.client
        )

# Create provider
provider = HanzoModelProvider(
    base_url="http://localhost:4000/v1",
    api_key="your-router-api-key"
)

# Use with agents
agent = Agent(name="Assistant", instructions="Be helpful and concise.")
result = await Runner.run(
    agent,
    "Hello!",
    run_config=RunConfig(model_provider=provider)
)

Supported Models

The Hanzo Router supports all major LLM providers:

OpenAI Models

  • gpt-4, gpt-4-turbo, gpt-4o
  • gpt-3.5-turbo
  • text-embedding-3-small, text-embedding-3-large

Anthropic Models

  • claude-3-opus-20240229
  • claude-3-sonnet-20240229
  • claude-3-haiku-20240307

Open Models

  • meta-llama/Llama-3-70b-chat-hf
  • mistralai/Mixtral-8x7B-Instruct-v0.1
  • google/gemma-7b-it
  • And many more...

Advanced Configuration

Custom Headers and Metadata

class CustomHanzoProvider(ModelProvider):
    def __init__(self, base_url: str, api_key: str, user_id: str = None):
        self.client = AsyncOpenAI(
            base_url=base_url,
            api_key=api_key,
            default_headers={
                "X-User-ID": user_id,
                "X-App-Name": "hanzo-agent-sdk"
            } if user_id else None
        )

Model Routing Strategies

# Use different models for different purposes
async def route_by_complexity(query: str) -> str:
    if len(query) < 50:  # Simple query
        model = "gpt-3.5-turbo"
    else:  # Complex query
        model = "gpt-4"
    
    result = await Runner.run(
        agent,
        query,
        run_config=RunConfig(
            model_provider=provider,
            model=model
        )
    )
    return result.final_output

Error Handling and Fallbacks

async def with_fallback(agent, query):
    providers = [
        ("gpt-4", primary_provider),
        ("claude-3-sonnet", fallback_provider),
        ("gpt-3.5-turbo", emergency_provider)
    ]
    
    for model, provider in providers:
        try:
            return await Runner.run(
                agent,
                query,
                run_config=RunConfig(
                    model_provider=provider,
                    model=model
                )
            )
        except Exception as e:
            print(f"Failed with {model}: {e}")
            continue
    
    raise Exception("All providers failed")

Testing Your Integration

Run the test script to verify your setup:

# From the agent SDK directory
python test_hanzo_backend.py

# Or run the example
python examples/hanzo_backend_example.py

Expected output:

=== Checking Hanzo Router Health ===
✓ Router health check passed
✓ Available models: 150 found

=== Testing Basic Agent ===
✓ Basic test passed!
Response: 4

=== Testing Agent with Tools ===
✓ Tool test passed!
Response: The result of 15 * 23 is 345.

Production Best Practices

1. API Key Management

Never hardcode API keys. Use environment variables or secret management:

import os
from dotenv import load_dotenv

load_dotenv()

api_key = os.getenv("HANZO_API_KEY")
if not api_key:
    raise ValueError("HANZO_API_KEY not found in environment")

2. Connection Pooling

Reuse the provider instance for better performance:

# Create once
_hanzo_provider = None

def get_hanzo_provider():
    global _hanzo_provider
    if _hanzo_provider is None:
        _hanzo_provider = HanzoModelProvider(
            base_url=os.getenv("HANZO_ROUTER_URL"),
            api_key=os.getenv("HANZO_API_KEY")
        )
    return _hanzo_provider

3. Monitoring and Logging

import logging

logger = logging.getLogger(__name__)

class MonitoredHanzoProvider(HanzoModelProvider):
    async def get_model(self, model_name: str | None) -> Model:
        start_time = time.time()
        try:
            model = await super().get_model(model_name)
            logger.info(f"Model {model_name} initialized in {time.time() - start_time:.2f}s")
            return model
        except Exception as e:
            logger.error(f"Failed to get model {model_name}: {e}")
            raise

4. Rate Limiting

The Router handles rate limiting, but you can add client-side controls:

from asyncio import Semaphore

class RateLimitedProvider(HanzoModelProvider):
    def __init__(self, *args, max_concurrent=10, **kwargs):
        super().__init__(*args, **kwargs)
        self._semaphore = Semaphore(max_concurrent)
    
    async def get_model(self, model_name: str | None) -> Model:
        async with self._semaphore:
            return await super().get_model(model_name)

Troubleshooting

Connection Errors

Cannot connect to Router: Cannot connect to host localhost:4000

Solution: Ensure the Router is running:

cd /path/to/hanzo/services
make start-router
# or
docker compose up router

Authentication Errors

401 Unauthorized: Invalid API key

Solution:

  1. Generate a new API key from the Router dashboard
  2. Update your environment variable: export HANZO_API_KEY="new-key"

Model Not Found

404: Model 'gpt-5' not found

Solution: Check available models:

curl http://localhost:4000/v1/models \
  -H "Authorization: Bearer $HANZO_API_KEY"

Integration with Other Hanzo Services

Using with MCP (Model Context Protocol)

# MCP tools can be integrated with agents
from hanzo_mcp import MCPClient

mcp_client = MCPClient()
tools = mcp_client.get_tools()

agent = Agent(
    name="MCPAgent",
    instructions="Use MCP tools to help users.",
    tools=tools
)

Observability with Cloud Dashboard

All requests through the Router are automatically logged to the Cloud dashboard:

  1. Access: http://localhost:3082 (local) or https://cloud.hanzo.ai (production)
  2. View real-time metrics, costs, and traces
  3. Set up alerts for errors or cost thresholds

Next Steps

Support