agent

README

Hanzo Agent SDK

A powerful Python framework for building AI agents and multi-agent systems with built-in orchestration.

✨ Features

• 🤖 Multi-Agent Networks: Build systems where multiple specialized agents collaborate
• 🧠 Intelligent Routing: Semantic, rule-based, and load-balanced routing strategies
• 🛠️ Powerful Tools: Enhanced tool system with MCP (Model Context Protocol) support
• 📊 Shared State: Agents can share information through network state
• 🔄 Orchestration: Define complex workflows with parallel, conditional, and loop steps
• 💾 Memory System: Long-term memory with vector search and reflection capabilities
• ⚡ UI Streaming: Real-time updates for building responsive interfaces
• 🔍 Observability: Built-in tracing and monitoring via Hanzo Cloud dashboard
• 🌐 Backend Flexibility: Use with Hanzo Router for 100+ LLM providers

Optional Extensions:

• 💎 Web3 Integration ([web3]): Wallet management, transactions, on-chain identity
• 🔒 TEE Support ([tee]): Intel SGX, AMD SEV, NVIDIA H100 attestation and confidential computing
• 🛒 Marketplace ([marketplace]): Decentralized agent service discovery and economics
• 💻 CLI ([cli]): Command-line interface integration

Core concepts:

1. Agents: LLMs configured with instructions, tools, and memory
2. Networks: Multi-agent systems with intelligent routing
3. Workflows: Orchestrate complex multi-step processes
4. State & Memory: Shared state and long-term memory
5. Tools: Enhanced tool system with MCP support
6. Tracing: Built-in tracking and observability

Explore the examples directory to see the SDK in action, and read our documentation for more details.

Notably, our SDK is compatible with any model providers that support the Open AI Chat Completions API format.

Get started

1. Set up your Python environment

python -m venv env
source env/bin/activate

1. Install Hanzo Agent SDK

# Basic installation
pip install hanzo-agent

# With Web3 support
pip install "hanzo-agent[web3]"

# With TEE support
pip install "hanzo-agent[tee]"

# With Marketplace support
pip install "hanzo-agent[marketplace]"

# With CLI support
pip install "hanzo-agent[cli]"

# Full installation (all extensions)
pip install "hanzo-agent[full]"

Quick Examples

Simple Agent

from agents import Agent, Runner

agent = Agent(name="Assistant", instructions="You are a helpful assistant")

result = Runner.run_sync(agent, "Write a haiku about recursion in programming.")
print(result.final_output)

# Code within the code,
# Functions calling themselves,
# Infinite loop's dance.

Multi-Agent Network

from agents import Agent
from agents.network import create_network, SemanticRouter

# Create specialized agents
researcher = Agent(
    name="Researcher",
    instructions="You find and analyze information.",
    tools=[search_tool, analyze_tool],
)

writer = Agent(
    name="Writer",
    instructions="You create content based on research.",
    tools=[format_tool],
)

# Create a network
network = create_network(
    agents=[researcher, writer],
    router=SemanticRouter(),
    default_model="gpt-4",
)

# Run the network
result = await network.run("Research and write about quantum computing")

Orchestrated Workflow

from agents import Agent
from agents.orchestration import Orchestrator, OrchestrationConfig

# Define agents (omitted: see Multi-Agent Network above)
researcher = Agent(name="researcher", instructions="...")
writer     = Agent(name="writer",     instructions="...")
reviewer   = Agent(name="reviewer",   instructions="...")

# Orchestrator owns the agent registry and workflow execution
orchestrator = Orchestrator(
    config=OrchestrationConfig(name="Content Pipeline"),
)
orchestrator.register_agent(researcher, capabilities=["research"])
orchestrator.register_agent(writer,     capabilities=["writing"])
orchestrator.register_agent(reviewer,   capabilities=["review"])

# Build a workflow from registered agents and execute it
workflow = orchestrator.create_workflow_from_agents(
    name="Article Workflow",
    agents=["researcher", "writer", "reviewer"],
)
orchestrator.register_workflow(workflow)

result = await orchestrator.execute_workflow(
    workflow_id=workflow.id,
    input="Research and write about AI safety.",
)
print(result.success, result.output)

For lower-level control — composing parallel, conditional, and loop steps directly — see agents.orchestration.Workflow plus the Step helpers (Step.agent, Step.parallel, Step.conditional, Step.loop).

(Configure backend with HANZO_ROUTER_URL and HANZO_API_KEY environment variables)

Handoffs example

from agents import Agent, Runner
import asyncio

spanish_agent = Agent(
    name="Spanish agent",
    instructions="You only speak Spanish.",
)

english_agent = Agent(
    name="English agent",
    instructions="You only speak English",
)

triage_agent = Agent(
    name="Triage agent",
    instructions="Handoff to the appropriate agent based on the language of the request.",
    handoffs=[spanish_agent, english_agent],
)


async def main():
    result = await Runner.run(triage_agent, input="Hola, ¿cómo estás?")
    print(result.final_output)
    # ¡Hola! Estoy bien, gracias por preguntar. ¿Y tú, cómo estás?


if __name__ == "__main__":
    asyncio.run(main())

Functions example

import asyncio

from agents import Agent, Runner, function_tool


@function_tool
def get_weather(city: str) -> str:
    return f"The weather in {city} is sunny."


agent = Agent(
    name="Hello world",
    instructions="You are a helpful agent.",
    tools=[get_weather],
)


async def main():
    result = await Runner.run(agent, input="What's the weather in Tokyo?")
    print(result.final_output)
    # The weather in Tokyo is sunny.


if __name__ == "__main__":
    asyncio.run(main())

The agent loop

When you call Runner.run(), we run a loop until we get a final output.

1. We call the LLM, using the model and settings on the agent, and the message history.
2. The LLM returns a response, which may include tool calls.
3. If the response has a final output (see below for more on this), we return it and end the loop.
4. If the response has a handoff, we set the agent to the new agent and go back to step 1.
5. We process the tool calls (if any) and append the tool responses messages. Then we go to step 1.

There is a max_turns parameter that you can use to limit the number of times the loop executes.

Final output

Final output is the last thing the agent produces in the loop.

1. If you set an output_type on the agent, the final output is when the LLM returns something of that type. We use structured outputs for this.
2. If there's no output_type (i.e. plain text responses), then the first LLM response without any tool calls or handoffs is considered as the final output.

As a result, the mental model for the agent loop is:

1. If the current agent has an output_type, the loop runs until the agent produces structured output matching that type.
2. If the current agent does not have an output_type, the loop runs until the current agent produces a message without any tool calls/handoffs.

Common agent patterns

The Agent SDK is designed to be highly flexible, allowing you to model a wide range of LLM workflows including deterministic flows, iterative loops, and more. See examples in examples/agent_patterns.

Tracing

The Agent SDK automatically traces your agent runs, making it easy to track and debug the behavior of your agents. Tracing is extensible by design, supporting custom spans and a wide variety of external destinations, including Logfire, AgentOps, Braintrust, Scorecard, and Keywords AI. For more details about how to customize or disable tracing, see Tracing.

Development (only needed if you need to edit the SDK/examples)

1. Ensure you have uv installed.

uv --version

1. Install dependencies

make sync

1. (After making changes) lint/test

make tests  # run tests
make mypy   # run typechecker
make lint   # run linter

Acknowledgements

We'd like to acknowledge the excellent work of the open-source community, especially:

• Pydantic (data validation) and PydanticAI (advanced agent framework)
• MkDocs
• Griffe
• uv and ruff

We're committed to continuing to build the Agent SDK as an open source framework so others in the community can expand on our approach.

Extension Examples

Web3 Agent

from agents import Agent, Runner
from agents.extensions.web3 import Web3Agent, AgentWallet

# Create wallet for agent
wallet = AgentWallet.from_mnemonic("your mnemonic here")

# Create Web3-enabled agent
trader = Web3Agent(
    name="Crypto Trader",
    instructions="You are a cryptocurrency trading agent",
    wallet=wallet
)

# Agent can now use wallet tools
result = await Runner.run(trader, "Check my ETH balance")

TEE (Confidential) Agent

from agents import Agent, Runner
from agents.extensions.tee import ConfidentialAgent, TEEProvider

# Create agent that runs in TEE
confidential_agent = ConfidentialAgent(
    name="Secure Agent",
    instructions="You handle sensitive data",
    tee_provider=TEEProvider.INTEL_SGX
)

# Generate attestation
attestation = await confidential_agent.generate_attestation()
print(f"Attestation: {attestation.quote}")

Marketplace Agent

from agents import Agent
from agents.extensions.marketplace import AgentMarketplace, ServiceOffer, ServiceType

# Create agent that offers services
service_agent = Agent(
    name="Research Agent",
    instructions="You conduct research"
)

# Create service offer
offer = ServiceOffer(
    id="research-1",
    agent_address="0x...",
    agent_name="Research Agent",
    service_type=ServiceType.RESEARCH,
    price_eth=0.01,
    description="Comprehensive research services"
)

# Register on marketplace
marketplace = AgentMarketplace()
await marketplace.register_offer(offer)

Documentation

Overview

Package agent is a reusable agentic-chat orchestrator: one POST runs a single LLM tool-calling round that lets a model manage a system through tools, and it PERSISTS conversation history (per-org SQLite via hanzoai/orm). It is a library — a host (hanzoai/ai / hanzoai/cloud) mounts it on its OWN zip router so the round registers in the SAME router as /v1/chat/completions (a distinct path, no route-precedence gamble), and injects the two dependency seams:

• Completer — the host's in-process LLM completion (the ONLY path that both returns tool_calls and carries per-org billing);
• ToolPlane — the host's unified tool registry (list / exists / dispatch).

A round yields four things: the assistant's text (reply), the tool calls the server executed against the registry (actions), the tool calls the client must apply itself — a graph/UI mutation the server cannot run (ops), and the id of the conversation the turn was appended to (conversationId).

This package DELIBERATELY does not import hanzoai/cloud or hanzoai/ai — that coupling is the thing it exists to break. It builds directly on zip (routing), hanzoai/orm (per-org SQLite persistence) and go-openai (the request/response shapes); the host wraps Mount with a thin adapter that supplies the seams.

Index

• Constants
• func Register(p Preset)
• type Completer
• type Conversation
• type Deps
• type Message
• type Preset
  • func Presets() []Preset
• type Principal
• type Scope
• type Service
  • func Mount(app *zip.App, deps Deps, completer Completer, plane ToolPlane) (*Service, error)
  • func MountAt(app *zip.App, prefix string, deps Deps, completer Completer, plane ToolPlane) (*Service, error)
  • func (s *Service) Close() error
• type Tool
• type ToolPlane
• type UpstreamError
  • func (e *UpstreamError) Error() string

Constants

const DefaultPrefix = "/v1/agent"

DefaultPrefix is where the standalone daemon answers. A host whose own router already spends that address folds this surface somewhere else with MountAt; nothing in the round depends on which address it was given.

const DefaultPreset = "graph"

DefaultPreset is used when a request omits preset (and its capability alias).

Functions

func Register

func Register(p Preset)

Register adds (or replaces) a preset in the library. Call from init() or before Mount. Replacing an existing id is allowed so a host can re-skin a builtin.

Types

type Completer

type Completer interface {
	Complete(ctx context.Context, cred map[string]string, req openai.ChatCompletionRequest) (openai.ChatCompletionResponse, error)
}

Completer runs one chat completion (with tools) and returns the parsed response. It is the seam onto the host's in-process LLM path: the real impl replays the request carrying the caller's credential (so the host's per-org billing runs); tests inject a fake so the round is exercised without a live model. agent never imports the package that provides the real Completer.

type Conversation

type Conversation struct {
	orm.Model[Conversation]
	Org   string `json:"org"`
	User  string `json:"user"`
	Title string `json:"title"`
}

Conversation is one persisted chat thread. Org is the owning org — physical isolation (one SQLite file per org) already scopes it; Org is stored for clarity and defense-in-depth. User is the member who opened it: a thread is one person's inside a shared org, so it lists and opens for them alone. A thread recorded before users were kept has none, and stays the org's.

type Deps

type Deps struct {
	// Logger is the canonical Hanzo logger (luxfi/log). Required.
	Logger luxlog.Logger
	// DataDir is the per-deployment data root. Per-org SQLite files land at
	// {DataDir}/orgs/{orgSlug}/agent.db. Required.
	DataDir string
	// Brand is the white-label brand identifier (logged only).
	Brand string
	// Model is the served model used when a request supplies none.
	Model string
	// Principal resolves the validated caller from a request. Defaults to
	// header-based resolution (X-Org-Id / X-User-Id) when nil.
	Principal func(*zip.Ctx) (Principal, bool)
}

Deps is agent's OWN small dependency surface — deliberately not cloud.Deps. It carries only what this package needs: a logger, the per-org SQLite data root, a brand tag, the default served model, and an optional principal resolver (defaults to gateway-header identity when nil).

type Message

type Message struct {
	orm.Model[Message]
	ConversationId string          `json:"conversationId"`
	Org            string          `json:"org"`
	Role           string          `json:"role"`
	Content        string          `json:"content"`
	ToolCalls      json.RawMessage `json:"toolCalls,omitempty"`
}

Message is one persisted turn. ConversationId is deliberately spelled with a lowercase-d so orm's PascalCase→camelCase filter (ToJSONFieldName lowercases only the first rune) maps Filter("ConversationId=") onto the stored "conversationId" JSON key. ToolCalls is the marshaled model tool_calls (nil for a plain user/assistant turn).

type Preset

type Preset struct {
	ID           string        `json:"id"`
	Title        string        `json:"title"`
	SystemPrompt string        `json:"systemPrompt"`
	BuiltinTools []openai.Tool `json:"-"`
	// ServerExecuted gates the tool-call split: when true, a call the tool
	// registry knows is dispatched server-side and reported in actions; when
	// false the round is advisory — every call is returned as an op for the
	// client to apply.
	ServerExecuted bool `json:"serverExecuted"`
}

Preset is one named agent type in the preset LIBRARY — a first-class, extensible catalog the round is created from and the presets route lists. It frames a tool-calling round: the system prompt that instructs the model, the builtin tool defs offered alongside the caller's and the org's registered tools, and whether the model's tool calls are EXECUTED server-side (registry Dispatch → actions) or handed back to the client as ops (a graph/UI mutation the server cannot perform). Adding an agent type is one Register call — no other change.

func Presets

func Presets() []Preset

Presets returns every registered preset, sorted by ID for a stable listing.

type Principal

type Principal struct {
	Org     string
	Project string
	User    string
	Cred    map[string]string
}

Principal is the VALIDATED caller a round runs as. Org scopes persistence AND the tool listing (never a client-supplied field); Cred is the caller's own credential headers, opaque to agent, replayed by the injected Completer / ToolPlane so an in-process call carries exactly the caller's identity.

type Scope

type Scope struct {
	Org     string
	Project string
}

Scope is the (org, project) a tool listing is resolved for.

type Service

type Service struct {
	// contains filtered or unexported fields
}

Service is the mounted orchestrator handle. Its handlers are plain zip handlers; Close releases the per-org SQLite stores at host shutdown.

func Mount

func Mount(app *zip.App, deps Deps, completer Completer, plane ToolPlane) (*Service, error)

Mount registers the routes at DefaultPrefix. See MountAt.

func MountAt

func MountAt(app *zip.App, prefix string, deps Deps, completer Completer, plane ToolPlane) (*Service, error)

MountAt registers the routes under prefix using the injected Completer and ToolPlane, and returns the Service so the host can Close it on shutdown. The per-org SQLite models are auto-migrated (schema created) on first per-org open. prefix is an absolute path chosen by whoever composes the router:

POST {prefix}                     — run one tool-calling round
GET  {prefix}/presets             — list the preset library
POST {prefix}/conversations       — record turns in a conversation
GET  {prefix}/conversations       — list the caller-org's conversations
GET  {prefix}/conversations/:id   — one conversation's messages

func (*Service) Close

func (s *Service) Close() error

Close releases every open per-org store. Idempotent.

type Tool

type Tool struct {
	Name         string
	Description  string
	Schema       json.RawMessage
	Activated    bool
	Dispatchable bool
}

Tool is the agent-facing projection of one registered tool offered to the model. Schema is the JSON-Schema of the call arguments; Dispatchable is false for a listing-only entry; Activated is true only for tools the org turned on.

type ToolPlane

type ToolPlane interface {
	List(ctx context.Context, scope Scope) []Tool
	Exists(ctx context.Context, scope Scope, name string) bool
	Dispatch(c *zip.Ctx, name string, args map[string]any) (any, error)
}

ToolPlane is the org's registered tool registry seam: list the tools offered to the model, test whether a call is server-known, and dispatch a call. Dispatch takes the live request so the host resolves the caller the ONE canonical way (its PrincipalFrom) — the credential is never reconstructed or passed as a value, only read from the validated request. List/Exists take a plain (org, project) Scope; that carries no credential, so it is safe by value. The host injects its unified tool plane; tests inject a stub.

type UpstreamError

type UpstreamError struct {
	Status int
	Body   []byte
}

UpstreamError is returned by a Completer when the completion service refuses the request for the CALLER's OWN reason — a 4xx like 402 insufficient_balance, 429 rate-limit, or 403 — that must reach the caller intact, not be masked as a gateway 502. Body is the completion's verbatim error payload; the round passes Status + Body straight through so a client shows the real message ("add credits") instead of an opaque "agent: completion" wrapper. A non-4xx completion failure (5xx, transport) stays a 502 — that IS a gateway fault.

func (*UpstreamError) Error

func (e *UpstreamError) Error() string