Zonix

Call simply. Chain deeply. Trace everything.

Zonix is a Python agent and workflow framework. It borrows the clarity of pydantic-ai's Agent, then adds workflow, team, and router primitives on top of one shared execution model. Agents, workflows, and teams are all nodes with the same __call__ / run / stream surface, and they share trace, usage, messages, and approval state.

The design goal is that a beginner can call one object and get a useful answer, while an advanced user can turn on reasoning, usage accounting, raw provider payloads, graph export, human approval, and frontend streaming without changing the shape of their business code.

Features

• One execution model. await node(task) returns the output, node.run(task) returns the full RunResult, node.stream(task) yields typed events. Agents, workflows, and teams all support the same three.
• Typed structured output. Set output=SomeModel and get a validated instance back, with automatic repair rounds when the model returns bad JSON.
• Tools from type hints. Schemas are generated from signatures and docstrings. Optional ToolContext injection, parallel execution, error capture, and middleware interception.
• Human approval built in. Mark a tool approval=True and run() returns a paused result you can resume(), or register an approver callback and keep the run going.
• Approval while streaming. Pass approval= to stream() and a pause is resolved by your handler mid-flight — the event stream keeps flowing instead of being cancelled.
• Composable orchestration. workflow gives then/parallel/join/ branch/loop, plus plain functions as steps via map; team gives router-driven dispatch. Both export Mermaid, DOT, SVG, PNG, or PDF graphs, and both can share message context across members.
• Resumable workflows. workflow(...).checkpoint(store) persists each step's result; rerun with the same run_id and finished steps are skipped. FileCheckpointStore and MemoryCheckpointStore ship in the box, or implement the Checkpointer protocol yourself.
• Explicit cancellation. Every entry point takes cancel= (an asyncio.Event, or anything with is_set()), checked between workflow steps, team turns, model calls, and tool calls.
• Provider-neutral adapters. OpenAI (Chat and Responses), Anthropic, and Gemini, plus offline Echo/StaticModel/ScriptedModel for tests. Any OpenAI-compatible endpoint works by setting base_url.
• Inspectable by default. RunResult keeps the span tree, usage, messages, and the raw upstream request and response for every model call.
• Sync facade. call_sync, run_sync, stream_sync for scripts, CLIs, and notebooks — including resumable approvals.

Install

pip install zonix

Optional provider extras:

pip install "zonix[openai]"
pip install "zonix[anthropic]"
pip install "zonix[gemini]"
pip install "zonix[viz]"     # image export for workflow/team graphs

Requires Python 3.11+. For local development from this repository: pip install -e .

60 seconds

import asyncio
import os

from pydantic import BaseModel

from zonix import agent
from zonix.models import OpenAI


class Plan(BaseModel):
    goal: str
    files: list[str]
    steps: list[str]


planner = agent(
    "planner",
    role="Plan code work",
    model=OpenAI("gpt-5.5", api_key=os.environ["OPENAI_API_KEY"]),
    output=Plan,
)


@planner.tool
def read_tree(path: str) -> list[str]:
    """List files under a repository path."""
    return sorted(os.listdir(path))


async def main() -> None:
    plan = await planner("add captcha to the login page")
    print(plan.goal, plan.files)

    result = await planner.run("add captcha to the login page")
    print(result.usage.total_tokens)
    print(result.model_calls[-1].raw_response)

    async for event in planner.stream("add captcha to the login page"):
        print(event)


asyncio.run(main())

Any OpenAI-compatible endpoint works with the same adapter — only the model name and base_url change:

from zonix.models import OpenAI

deepseek = OpenAI(
    model="deepseek-chat",
    api_key=os.environ["DEEPSEEK_API_KEY"],
    base_url="https://api.deepseek.com/v1",
)

Streaming chat requests automatically send stream_options={"include_usage": true}, so token counts are complete even on streamed runs.

Multi-agent

from zonix import router, team, workflow
from zonix.types import Route

# Fixed pipeline: output of one step feeds the next.
flow = (
    workflow("review")
    .start(planner)
    .parallel(security_review, perf_review)
    .join(merge_reviews)
    .branch(lambda r: r.risk == "high", then=human_gate, else_=auto_apply)
    .build()
)

review = await flow("audit the auth changes", ctx=ctx)
print(flow.to_mermaid())


# Router-driven dispatch: the router picks the next node each step.
def choose(task, state) -> Route:
    if isinstance(task, Review):
        return Route(done=True)
    return Route(next="reviewer" if "review" in str(task).lower() else "coder")


code_team = (
    team("code_team")
    .add(planner, coder, reviewer)
    .route(router("rule_router", choose))
    .build(max_steps=6)
)

answer = await code_team("review the auth changes", ctx=ctx)

A router can be a rule function, another agent, or any node that returns Route(next=..., done=..., input=...). Workflows and teams are nodes themselves, so they nest freely.

Long pipelines can checkpoint each step and resume where they stopped:

from zonix import FileCheckpointStore

flow = (
    workflow("review")
    .start(planner)
    .map(lambda plan: plan.model_dump())    # plain functions are steps too
    .then(coder)
    .checkpoint(FileCheckpointStore("./.checkpoints"))
    .share_context()                        # each step sees the ones before it
    .build()
)

await flow("audit the auth changes", run_id="job-42")   # rerun skips finished steps

Checkpointed values round-trip as JSON, so a step that returned a BaseModel replays as a dict — keep plain jsonable values flowing between steps, or re-validate at the start of the next one.

Human approval

result = await coder.run("edit the login page", ctx=ctx)

if result.paused:
    print(result.pending.tool, result.pending.input)
    result = await result.resume(approve=True)

Or skip the pause entirely by passing an approver:

result = await coder.run("edit the login page", approval=lambda pending: True)

Paused results hold a live continuation. Release them with await result.cancel() (or result.close() for run_sync) if you will not resume.

Streaming takes the same handler, and the stream survives the pause:

async for event in coder.stream("edit the login page", approval=my_handler):
    ...

Without a handler a streamed run emits ApprovalRequired and then cancels.

Cancellation

cancel = asyncio.Event()
task = asyncio.create_task(flow.run("long job", cancel=cancel))
cancel.set()     # cooperative: takes effect at the next checkpoint

Tracing

Zonix ships vendor-neutral tracing hooks with no collector dependency. Register a SpanProcessor once, then override per run:

from zonix import TraceOptions, configure_tracing

configure_tracing(my_processor, defaults=TraceOptions(enabled=True, project="my-app"))

result = await planner.run(task, trace=TraceOptions(tags=["dev"], metadata={"user_id": "u1"}))
async for event in planner.stream(task, trace=TraceOptions(tags=["ui"])):
    ...

The span tree covers workflows, teams, agents, routers, model calls, and tool calls. result.trace stays available even when export is disabled. The separate zonix-observe package provides a local collector, storage, and browser UI.

Architecture

zonix/
  spec.py       agent()/team()/workflow()/router() factories
  engine.py     agent model and tool execution loop
  runtime.py    __call__/run/stream driver shared by every node
  types.py      Message, Usage, Span, RunState, RunResult, Route
  tools.py      tool definitions, ToolContext, middleware results
  graph.py      graph specs, Mermaid, DOT, and image export
  checkpoint.py Checkpointer protocol, file and in-memory stores
  memory/       Window, Summarize, Vector, Session
  multi/        Workflow, Team, Router nodes
  models/       OpenAI, Anthropic, Gemini, offline adapters
  hitl.py       approval keys and snapshot persistence
  tracing.py    vendor-neutral span lifecycle and processor hooks
  wire/         event-to-wire protocol adapters (Vercel AI SDK)

Documentation

• 中文完整 API 教程
• Single agent example
• Workflow and team example
• Real provider smoke script

License

MIT