mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00

Files

T

zeekayandHanzo Dev 252b839072 hanzo-tasks speaks to the engine we actually run

0.1.0 shipped a Temporal gRPC client under our name: it depended on
temporalio and dialled localhost:7233, a port the native engine does not
serve. Anyone who installed it got a client that could not reach us.

The engine (hanzoai/tasks) is native and speaks JSON over
/v1/tasks/namespaces/{ns}/activities. So this drops temporalio for httpx
and rewrites the package against that surface: a client that dispatches,
reads and settles activities, and a worker that claims them.

The worker POLLS rather than accepting a push, which is what lets it run
behind NAT — the claim endpoint exists for exactly that, and the shipped
Go worker (`hanzo gpu connect`) already pulls from it.

The engine's rules are not copied here. It reaps expired leases before
every claim, serializes claims per namespace, and derives the lease from
the activity's own heartbeat timeout; a second opinion on any of that
could only disagree with it. What the worker does hold up is the lease
while a handler runs, beating at a third of the window the server
granted — and a plain function runs in a thread, because a blocking call
on the event loop would stall that beat and let the work be reaped
out from under itself.

The Temporal workflow modules are deleted rather than ported. Replay
determinism is not something this surface offers, and a @workflow.defn
that cannot replay would be a promise the engine never made.

Also names the sdist contents, so a local virtualenv beside the package
can never be swept into a release again.

Co-authored-by: Hanzo Dev <[email protected]>

2026-08-06 15:05:32 -07:00

3.0 KiB

Raw Permalink Blame History

hanzo-tasks

Durable activities on the native Hanzo Tasks engine.

The engine is Hanzo's own (hanzoai/tasks), reached over JSON at api.hanzo.ai/v1/tasks. It speaks no protobuf and no gRPC, and this package depends on no workflow runtime — dispatching work and running a worker are both plain HTTP over httpx.

pip install hanzo-tasks

Dispatch

from hanzo_tasks import Tasks

async with Tasks.from_env() as tasks:
    activity = await tasks.dispatch(
        "render",
        input={"scene": 3},
        task_queue="gpu",
        heartbeat_timeout="60s",
    )
    print(activity.id, activity.run_id, activity.status)

from_env() reads HANZO_API_KEY, HANZO_TASKS_URL and HANZO_TASKS_NAMESPACE. The org is never named in a request: IAM validates the bearer token and the edge mints the org from the validated claim, so a caller cannot reach another tenant's shard by asking to.

Run a worker

The worker polls. It never accepts a push, so it needs no inbound address and runs behind NAT.

from hanzo_tasks import Tasks, Worker

tasks = Tasks.from_env()
worker = Worker(tasks, task_queue="gpu")


@worker.handler("render")
async def render(activity):
    return {"frames": 120}


await worker.run()

Return a value to complete the activity; raise to fail it with that cause. Handlers may be async or plain functions.

What the engine does, so you don't

These are properties of the engine, not of this client. There is deliberately no second copy of them here, because a second copy can disagree.

A dead worker's work comes back. The engine reaps expired leases before every claim, so an activity whose claimant stopped heartbeating returns to SCHEDULED and the next poll picks it up. No client-side timer.
Two workers never claim the same activity. Claims are serialized per namespace.
The lease is the activity's own heartbeat timeout, else the worker's lease_seconds, else the engine default. Worker heartbeats at a third of the window the server actually granted, so a beat can be missed without losing the claim.
An empty queue is not an error. It is 204, and claim() returns None.
Retries are the engine's. Pass a RetryPolicy to dispatch(); the engine counts attempts and fails the activity when they are exhausted.

Reading state

page = await tasks.activities(page_size=50)
for activity in page:
    print(activity.id, activity.status, activity.attempt)

activity = await tasks.describe(id, run_id)
events, cursor = await tasks.history(id, run_id)

Errors

The engine reports {"error": ..., "code": <int>} — code is a number here, not the status string the rest of the Hanzo API uses.

Exception	Meaning
`Denied`	401/403 — no validated principal, or one carrying no org
`NotFound`	404 — no such activity in this namespace
`Terminal`	409 — already completed, failed or canceled
`TasksError`	anything else, carrying `.code`

License

MIT

3.0 KiB Raw Permalink Blame History