Hanzo Python SDK

# pip
pip install hanzoai

# uv (preferred)
uv add hanzoai

# poetry
poetry add hanzoai

from hanzoai import Hanzo

client = Hanzo(
    api_key="your-api-key",  # defaults to HANZO_API_KEY env var
)

response = client.chat.completions.create(
    model="zen-70b",
    messages=[{"role": "user", "content": "Hello, Hanzo!"}],
)
print(response.choices[0].message.content)

stream = client.chat.completions.create(
    model="zen-70b",
    messages=[{"role": "user", "content": "Write a poem about code"}],
    stream=True,
)

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

import asyncio
from hanzoai import AsyncHanzo

client = AsyncHanzo()

async def main():
    response = await client.chat.completions.create(
        model="zen-70b",
        messages=[{"role": "user", "content": "Hello!"}],
    )
    print(response.choices[0].message.content)

asyncio.run(main())

async def stream_response():
    stream = await client.chat.completions.create(
        model="zen-70b",
        messages=[{"role": "user", "content": "Write a story"}],
        stream=True,
    )

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

embedding = client.embeddings.create(
    model="zen-embedding",
    input="Hello world",
)
print(len(embedding.data[0].embedding))  # dimension count

response = client.chat.completions.create(
    model="zen-70b",
    messages=[{"role": "user", "content": "What's the weather in Tokyo?"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {"type": "string", "description": "City name"},
                },
                "required": ["location"],
            },
        },
    }],
)

tool_call = response.choices[0].message.tool_calls[0]
if tool_call:
    print(tool_call.function.name)       # "get_weather"
    print(tool_call.function.arguments)  # '{"location":"Tokyo"}'

response = client.chat.completions.create(
    model="zen-vision",
    messages=[{
        "role": "user",
        "content": [
            {"type": "text", "text": "What's in this image?"},
            {"type": "image_url", "image_url": {"url": "https://example.com/photo.jpg"}},
        ],
    }],
)

client = Hanzo(
    api_key="your-key",                    # Required (or HANZO_API_KEY env)
    base_url="https://api.hanzo.ai",       # Default
    timeout=60.0,                          # Request timeout (seconds)
    max_retries=2,                         # Auto-retry count
    default_headers={"X-Custom": "val"},   # Extra headers
    default_query={"version": "2"},        # Extra query params
)

HANZO_API_KEY=your-api-key          # Required
HANZO_BASE_URL=https://...          # Override base URL
HANZO_LOG=debug                     # Enable debug logging

from openai import OpenAI

client = OpenAI(
    base_url="https://api.hanzo.ai/v1",
    api_key=os.environ["HANZO_API_KEY"],
)
# Everything works — same API shape
response = client.chat.completions.create(
    model="zen-70b",
    messages=[{"role": "user", "content": "Hello"}],
)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from hanzoai import Hanzo

app = FastAPI()
client = Hanzo()

@app.post("/api/chat")
async def chat(messages: list[dict]):
    stream = client.chat.completions.create(
        model="zen-70b",
        messages=messages,
        stream=True,
    )

    async def generate():
        for chunk in stream:
            content = chunk.choices[0].delta.content
            if content:
                yield f"data: {content}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    base_url="https://api.hanzo.ai/v1",
    api_key=os.environ["HANZO_API_KEY"],
    model="zen-70b",
)

response = llm.invoke("Hello, Hanzo!")

from hanzoai import Hanzo, APIError, RateLimitError, APIConnectionError

try:
    response = client.chat.completions.create(...)
except RateLimitError as e:
    print(f"Rate limited, retry after: {e.response.headers.get('retry-after')}")
except APIError as e:
    print(f"API error: {e.status_code} {e.message}")
except APIConnectionError as e:
    print(f"Connection error: {e}")

# Auto-pagination
for model in client.models.list():
    print(model.id)

# Manual pagination
page = client.models.list()
print(page.data)
if page.has_next_page():
    next_page = page.get_next_page()

hanzoai/
├── __init__.py           # Main client export
├── _client.py            # Hanzo and AsyncHanzo clients
├── _streaming.py         # Stream and AsyncStream
├── _response.py          # Response types
├── types/                # All Pydantic response models
│   ├── chat/
│   ├── fine_tuning/
│   ├── audio/
│   └── ...
├── resources/            # Resource namespace implementations
│   ├── chat/
│   │   └── completions.py
│   ├── embeddings.py
│   ├── models.py
│   ├── files.py
│   └── ...
└── pkg/                  # 55 sub-packages