mirror of
https://github.com/hanzoai/python-sdk.git
synced 2026-10-23 20:23:00 +00:00
hanzo-tasks: the client is the /v1/tasks contract, not a Temporal wrapper
The package dialed Temporal over gRPC, a transport hanzoai/tasks removed in
v3, so nothing it did could reach the engine. It is now the HTTP contract the
engine serves under /v1/tasks/namespaces/{ns}: start, signal-with-start, list,
describe, signal, query, cancel, terminate, reset, history, and schedules — one
method per route, IAM bearer, tested against the request shapes. A workflow's
body still runs in a Go or TypeScript worker.
Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
Hanzo Dev
@@ -0,0 +1,26 @@
|
||||
# hanzo-tasks
|
||||
|
||||
The Hanzo Tasks client: durable workflows over `/v1/tasks`.
|
||||
|
||||
Hanzo Cloud embeds the hanzoai/tasks engine and serves it at
|
||||
`https://api.hanzo.ai/v1/tasks/namespaces/{namespace}/…`, authenticated by your
|
||||
IAM bearer. This package is that HTTP contract, one method per route. A
|
||||
workflow's body runs in a worker written with the Go (`tasks/pkg/sdk`) or
|
||||
TypeScript (`@hanzoai/tasks`) SDK.
|
||||
|
||||
```bash
|
||||
pip install hanzo-tasks
|
||||
```
|
||||
|
||||
```python
|
||||
from hanzo_tasks import Client, TasksConfig
|
||||
|
||||
with Client(TasksConfig(token=TOKEN, namespace="acme")) as tasks:
|
||||
wf = tasks.start("agent.review", task_queue="agents", input={"repo": "acme/widgets"})
|
||||
tasks.signal(wf.workflow_id, "approve", {"by": "z"})
|
||||
state = tasks.query(wf.workflow_id, "status")
|
||||
tasks.cancel(wf.workflow_id, reason="superseded")
|
||||
```
|
||||
|
||||
Routes covered: start, signal-with-start, list, describe, signal, query,
|
||||
cancel, terminate, reset, history; schedules create/get/list/delete/trigger.
|
||||
@@ -1,43 +1,18 @@
|
||||
"""
|
||||
hanzo-tasks — Durable workflow execution for AI agents.
|
||||
"""hanzo-tasks — the Hanzo Tasks client.
|
||||
|
||||
Wraps the Temporal Python SDK with Hanzo conventions for agent task
|
||||
orchestration, including pre-built workflows for pipelines and fan-out.
|
||||
Durable workflows in Hanzo Cloud run on hanzoai/tasks, embedded in the cloud
|
||||
process and served under `/v1/tasks`. From Python you start, signal, query,
|
||||
cancel, terminate and reset them, and manage schedules; a workflow's body runs
|
||||
in a worker written with the Go (`tasks/pkg/sdk`) or TypeScript
|
||||
(`@hanzoai/tasks`) SDK.
|
||||
|
||||
Example:
|
||||
>>> from hanzo_tasks import Client, TasksConfig
|
||||
>>> client = await Client.connect(TasksConfig(namespace="hanzo"))
|
||||
>>> handle = await client.submit(AgentTaskWorkflow.run, task_input, queue="agents")
|
||||
>>> result = await handle.result()
|
||||
>>> with Client(TasksConfig(token=TOKEN, namespace="acme")) as tasks:
|
||||
... wf = tasks.start("agent.review", task_queue="agents", input={"repo": "acme/widgets"})
|
||||
... tasks.signal(wf.workflow_id, "approve", {"by": "z"})
|
||||
"""
|
||||
|
||||
from .activities import execute_agent_task, send_notification, set_agent_executor
|
||||
from .client import Client, TasksConfig, WorkflowHandle
|
||||
from .worker import Worker
|
||||
from .workflows import (
|
||||
AgentTaskInput,
|
||||
AgentTaskOutput,
|
||||
AgentTaskWorkflow,
|
||||
FanOutWorkflow,
|
||||
PipelineWorkflow,
|
||||
)
|
||||
from .client import DEFAULT_HOST, Client, TasksConfig, TasksError, Workflow
|
||||
|
||||
__version__ = "0.1.0"
|
||||
__all__ = [
|
||||
# Client
|
||||
"Client",
|
||||
"TasksConfig",
|
||||
"WorkflowHandle",
|
||||
# Worker
|
||||
"Worker",
|
||||
# Workflows
|
||||
"AgentTaskWorkflow",
|
||||
"PipelineWorkflow",
|
||||
"FanOutWorkflow",
|
||||
"AgentTaskInput",
|
||||
"AgentTaskOutput",
|
||||
# Activities
|
||||
"execute_agent_task",
|
||||
"send_notification",
|
||||
"set_agent_executor",
|
||||
]
|
||||
__version__ = "0.2.0"
|
||||
__all__ = ["Client", "TasksConfig", "TasksError", "Workflow", "DEFAULT_HOST"]
|
||||
|
||||
@@ -1,35 +0,0 @@
|
||||
"""Activity definitions for agent task execution."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Callable
|
||||
|
||||
from temporalio import activity
|
||||
|
||||
# Activity executor — pluggable. The playground sets this to call the ZAP sidecar.
|
||||
_agent_executor: Callable[..., Any] | None = None
|
||||
|
||||
|
||||
def set_agent_executor(fn: Callable[..., Any]) -> None:
|
||||
"""Set the function that executes agent tasks (called by playground)."""
|
||||
global _agent_executor
|
||||
_agent_executor = fn
|
||||
|
||||
|
||||
@activity.defn
|
||||
async def execute_agent_task(input: Any) -> Any:
|
||||
"""Execute an agent task. Delegates to the registered executor."""
|
||||
if _agent_executor is None:
|
||||
raise RuntimeError(
|
||||
"No agent executor registered. Call set_agent_executor() first."
|
||||
)
|
||||
return await _agent_executor(input)
|
||||
|
||||
|
||||
@activity.defn
|
||||
async def send_notification(input: dict[str, Any]) -> None:
|
||||
"""Send a notification (webhook, SSE, etc)."""
|
||||
import httpx
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
await client.post(input.get("url", ""), json=input)
|
||||
@@ -1,100 +1,237 @@
|
||||
"""Hanzo Tasks client — wraps Temporal client with Hanzo conventions."""
|
||||
"""The Hanzo Tasks client: `/v1/tasks` over HTTP.
|
||||
|
||||
The durable engine (hanzoai/tasks) is embedded in Hanzo Cloud and answers under
|
||||
`/v1/tasks/namespaces/{namespace}/...` on the same origin as everything else,
|
||||
authenticated by the caller's IAM bearer. A workflow is started, signalled,
|
||||
queried, cancelled, terminated and reset here; its body runs in a worker
|
||||
written with the Go or TypeScript SDK. This module holds no engine protocol of
|
||||
its own — it is the HTTP contract, one method per route.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import uuid
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
from uuid import uuid4
|
||||
|
||||
from temporalio.client import Client as TemporalClient
|
||||
import httpx
|
||||
|
||||
DEFAULT_HOST = "https://api.hanzo.ai"
|
||||
|
||||
|
||||
@dataclass
|
||||
@dataclass(frozen=True)
|
||||
class TasksConfig:
|
||||
"""Configuration for connecting to Temporal."""
|
||||
"""Where the engine is and who is calling.
|
||||
|
||||
address: str = "localhost:7233"
|
||||
namespace: str = "hanzo"
|
||||
tls: bool = False
|
||||
`token` is an IAM bearer (a user or service token); the org is read from it
|
||||
server-side. `namespace` is the engine namespace the caller's workflows live
|
||||
in — the org's own by convention.
|
||||
"""
|
||||
|
||||
token: str
|
||||
namespace: str = "default"
|
||||
host: str = DEFAULT_HOST
|
||||
timeout: float = 30.0
|
||||
|
||||
|
||||
class WorkflowHandle:
|
||||
"""Handle to a running workflow."""
|
||||
@dataclass(frozen=True)
|
||||
class Workflow:
|
||||
"""One workflow execution as the engine describes it."""
|
||||
|
||||
def __init__(self, handle: Any) -> None:
|
||||
self._handle = handle
|
||||
workflow_id: str
|
||||
run_id: str
|
||||
status: str = ""
|
||||
workflow_type: str = ""
|
||||
raw: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
@property
|
||||
def id(self) -> str:
|
||||
return self._handle.id
|
||||
@classmethod
|
||||
def of(cls, d: dict[str, Any]) -> "Workflow":
|
||||
wt = d.get("workflowType")
|
||||
return cls(
|
||||
workflow_id=str(d.get("workflowId", "")),
|
||||
run_id=str(d.get("runId", "")),
|
||||
status=str(d.get("status", "")),
|
||||
workflow_type=str(wt.get("name", "") if isinstance(wt, dict) else wt or ""),
|
||||
raw=d,
|
||||
)
|
||||
|
||||
@property
|
||||
def run_id(self) -> str:
|
||||
return self._handle.result_run_id
|
||||
|
||||
async def result(self, result_type: type | None = None) -> Any:
|
||||
return await self._handle.result(result_type=result_type)
|
||||
class TasksError(RuntimeError):
|
||||
"""The engine refused or failed a call; `status` is the HTTP status."""
|
||||
|
||||
async def cancel(self) -> None:
|
||||
await self._handle.cancel()
|
||||
|
||||
async def signal(self, name: str, data: Any = None) -> None:
|
||||
await self._handle.signal(name, data)
|
||||
def __init__(self, status: int, message: str) -> None:
|
||||
super().__init__(f"tasks: HTTP {status}: {message}")
|
||||
self.status = status
|
||||
|
||||
|
||||
class Client:
|
||||
"""Hanzo Tasks client for submitting and managing workflows."""
|
||||
"""Synchronous client over `/v1/tasks`. Use as a context manager or call `close()`."""
|
||||
|
||||
def __init__(self, temporal: TemporalClient) -> None:
|
||||
self._temporal = temporal
|
||||
|
||||
@classmethod
|
||||
async def connect(cls, config: TasksConfig | None = None) -> Client:
|
||||
"""Connect to Temporal server."""
|
||||
cfg = config or TasksConfig()
|
||||
temporal = await TemporalClient.connect(cfg.address, namespace=cfg.namespace)
|
||||
return cls(temporal)
|
||||
|
||||
async def submit(
|
||||
self,
|
||||
workflow: Any,
|
||||
input: Any,
|
||||
*,
|
||||
id: str | None = None,
|
||||
queue: str = "default",
|
||||
**kwargs: Any,
|
||||
) -> WorkflowHandle:
|
||||
"""Submit a workflow for execution."""
|
||||
handle = await self._temporal.start_workflow(
|
||||
workflow,
|
||||
input,
|
||||
id=id or str(uuid4()),
|
||||
task_queue=queue,
|
||||
**kwargs,
|
||||
def __init__(self, config: TasksConfig, transport: httpx.BaseTransport | None = None) -> None:
|
||||
self.config = config
|
||||
self._http = httpx.Client(
|
||||
base_url=config.host.rstrip("/"),
|
||||
headers={"Authorization": f"Bearer {config.token}", "Content-Type": "application/json"},
|
||||
timeout=config.timeout,
|
||||
transport=transport,
|
||||
)
|
||||
return WorkflowHandle(handle)
|
||||
|
||||
async def get_result(self, workflow_id: str, result_type: type | None = None) -> Any:
|
||||
"""Get the result of a completed workflow."""
|
||||
handle = self._temporal.get_workflow_handle(workflow_id)
|
||||
return await handle.result(result_type=result_type)
|
||||
# ── plumbing ─────────────────────────────────────────────────────────────
|
||||
|
||||
async def cancel(self, workflow_id: str) -> None:
|
||||
"""Cancel a running workflow."""
|
||||
handle = self._temporal.get_workflow_handle(workflow_id)
|
||||
await handle.cancel()
|
||||
def _path(self, *parts: str) -> str:
|
||||
return "/v1/tasks/namespaces/" + "/".join([self.config.namespace, *parts])
|
||||
|
||||
async def signal(self, workflow_id: str, signal_name: str, data: Any = None) -> None:
|
||||
"""Send a signal to a running workflow."""
|
||||
handle = self._temporal.get_workflow_handle(workflow_id)
|
||||
await handle.signal(signal_name, data)
|
||||
def _call(self, method: str, path: str, *, params: dict[str, Any] | None = None, body: Any = None) -> Any:
|
||||
r = self._http.request(method, path, params=params or None, json=body)
|
||||
if r.status_code >= 400:
|
||||
try:
|
||||
msg = r.json().get("error") or r.text
|
||||
except ValueError:
|
||||
msg = r.text
|
||||
raise TasksError(r.status_code, str(msg).strip())
|
||||
if not r.content:
|
||||
return None
|
||||
try:
|
||||
return r.json()
|
||||
except ValueError:
|
||||
return r.text
|
||||
|
||||
async def query(self, workflow_id: str, query_name: str) -> Any:
|
||||
"""Query a running workflow."""
|
||||
handle = self._temporal.get_workflow_handle(workflow_id)
|
||||
return await handle.query(query_name)
|
||||
def close(self) -> None:
|
||||
self._http.close()
|
||||
|
||||
@property
|
||||
def temporal(self) -> TemporalClient:
|
||||
"""Access the underlying Temporal client."""
|
||||
return self._temporal
|
||||
def __enter__(self) -> "Client":
|
||||
return self
|
||||
|
||||
def __exit__(self, *_: object) -> None:
|
||||
self.close()
|
||||
|
||||
# ── workflows ────────────────────────────────────────────────────────────
|
||||
|
||||
def start(
|
||||
self,
|
||||
workflow_type: str,
|
||||
*,
|
||||
task_queue: str,
|
||||
input: Any = None,
|
||||
workflow_id: str | None = None,
|
||||
request_id: str | None = None,
|
||||
) -> Workflow:
|
||||
"""Start a workflow. A caller-chosen `workflow_id` makes the start
|
||||
idempotent; `request_id` deduplicates a retried request."""
|
||||
body = {
|
||||
"workflowId": workflow_id or str(uuid.uuid4()),
|
||||
"workflowType": {"name": workflow_type},
|
||||
"taskQueue": {"name": task_queue},
|
||||
"input": input,
|
||||
"requestId": request_id or str(uuid.uuid4()),
|
||||
}
|
||||
return Workflow.of(self._call("POST", self._path("workflows"), body=body) or body)
|
||||
|
||||
def signal_with_start(
|
||||
self,
|
||||
workflow_type: str,
|
||||
*,
|
||||
task_queue: str,
|
||||
signal: str,
|
||||
payload: Any = None,
|
||||
input: Any = None,
|
||||
workflow_id: str | None = None,
|
||||
request_id: str | None = None,
|
||||
) -> Workflow:
|
||||
"""Signal a workflow, starting it first if it is not running."""
|
||||
body = {
|
||||
"workflowId": workflow_id or str(uuid.uuid4()),
|
||||
"workflowType": {"name": workflow_type},
|
||||
"taskQueue": {"name": task_queue},
|
||||
"input": input,
|
||||
"signalName": signal,
|
||||
"signalPayload": payload,
|
||||
"requestId": request_id or str(uuid.uuid4()),
|
||||
}
|
||||
return Workflow.of(self._call("POST", self._path("workflows", "signal-with-start"), body=body) or body)
|
||||
|
||||
def list(self, query: str = "") -> list[Workflow]:
|
||||
out = self._call("GET", self._path("workflows"), params={"query": query} if query else None)
|
||||
rows = out.get("workflows", out) if isinstance(out, dict) else out
|
||||
return [Workflow.of(w) for w in (rows or [])]
|
||||
|
||||
def describe(self, workflow_id: str, run_id: str = "") -> Workflow:
|
||||
out = self._call("GET", self._path("workflows", workflow_id), params={"runId": run_id} if run_id else None)
|
||||
return Workflow.of(out or {"workflowId": workflow_id, "runId": run_id})
|
||||
|
||||
def signal(self, workflow_id: str, name: str, payload: Any = None, run_id: str = "") -> None:
|
||||
self._call(
|
||||
"POST",
|
||||
self._path("workflows", workflow_id, "signal"),
|
||||
params={"runId": run_id} if run_id else None,
|
||||
body={"name": name, "payload": payload},
|
||||
)
|
||||
|
||||
def query(self, workflow_id: str, query_type: str, args: Any = None, run_id: str = "") -> Any:
|
||||
return self._call(
|
||||
"POST",
|
||||
self._path("workflows", workflow_id, "query"),
|
||||
params={"runId": run_id} if run_id else None,
|
||||
body={"queryType": query_type, "args": args},
|
||||
)
|
||||
|
||||
def cancel(self, workflow_id: str, reason: str = "", run_id: str = "", identity: str = "") -> Workflow:
|
||||
out = self._call(
|
||||
"POST",
|
||||
self._path("workflows", workflow_id, "cancel"),
|
||||
params={"runId": run_id} if run_id else None,
|
||||
body={"reason": reason, "identity": identity},
|
||||
)
|
||||
return Workflow.of(out or {"workflowId": workflow_id, "runId": run_id})
|
||||
|
||||
def terminate(self, workflow_id: str, reason: str = "", run_id: str = "", identity: str = "") -> Workflow:
|
||||
out = self._call(
|
||||
"POST",
|
||||
self._path("workflows", workflow_id, "terminate"),
|
||||
params={"runId": run_id} if run_id else None,
|
||||
body={"reason": reason, "identity": identity},
|
||||
)
|
||||
return Workflow.of(out or {"workflowId": workflow_id, "runId": run_id})
|
||||
|
||||
def reset(self, workflow_id: str, event_id: int, run_id: str = "", reason: str = "", identity: str = "") -> Workflow:
|
||||
out = self._call(
|
||||
"POST",
|
||||
self._path("workflows", workflow_id, "reset"),
|
||||
body={"runId": run_id, "eventId": event_id, "reason": reason, "identity": identity},
|
||||
)
|
||||
return Workflow.of(out or {"workflowId": workflow_id, "runId": run_id})
|
||||
|
||||
def history(self, workflow_id: str, run_id: str = "", after: int = 0, page_size: int = 0, reverse: bool = False) -> Any:
|
||||
params: dict[str, Any] = {}
|
||||
if run_id:
|
||||
params["runId"] = run_id
|
||||
if after:
|
||||
params["afterId"] = after
|
||||
if page_size:
|
||||
params["pageSize"] = page_size
|
||||
if reverse:
|
||||
params["reverse"] = "true"
|
||||
return self._call("GET", self._path("workflows", workflow_id, "history"), params=params)
|
||||
|
||||
# ── schedules ────────────────────────────────────────────────────────────
|
||||
|
||||
def schedules(self) -> Any:
|
||||
return self._call("GET", self._path("schedules"))
|
||||
|
||||
def create_schedule(self, schedule: dict[str, Any]) -> Any:
|
||||
"""Create a schedule; `schedule` is the engine's schedule document
|
||||
(id, spec, action) as the Go and TypeScript SDKs send it."""
|
||||
return self._call("POST", self._path("schedules"), body=schedule)
|
||||
|
||||
def schedule(self, schedule_id: str) -> Any:
|
||||
return self._call("GET", self._path("schedules", schedule_id))
|
||||
|
||||
def delete_schedule(self, schedule_id: str) -> None:
|
||||
self._call("DELETE", self._path("schedules", schedule_id))
|
||||
|
||||
def trigger_schedule(self, schedule_id: str, overlap_policy: str = "", request_id: str | None = None) -> Any:
|
||||
return self._call(
|
||||
"POST",
|
||||
self._path("schedules", schedule_id, "trigger"),
|
||||
body={"requestId": request_id or str(uuid.uuid4()), "overlapPolicy": overlap_policy},
|
||||
)
|
||||
|
||||
@@ -1,54 +0,0 @@
|
||||
"""Hanzo Tasks worker — polls and executes activities."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from temporalio.worker import Worker as TemporalWorker
|
||||
|
||||
|
||||
class Worker:
|
||||
"""Hanzo Tasks worker that polls a queue and executes workflows/activities."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
client: Any,
|
||||
queue: str = "default",
|
||||
workflows: list[Any] | None = None,
|
||||
activities: list[Any] | None = None,
|
||||
) -> None:
|
||||
self._client = client
|
||||
self._queue = queue
|
||||
self._workflows: list[Any] = workflows or []
|
||||
self._activities: list[Any] = activities or []
|
||||
self._worker: TemporalWorker | None = None
|
||||
|
||||
def register_workflow(self, workflow_cls: Any) -> Any:
|
||||
"""Register a workflow class. Can be used as a decorator."""
|
||||
self._workflows.append(workflow_cls)
|
||||
return workflow_cls
|
||||
|
||||
def register_activity(self, activity_fn: Any) -> Any:
|
||||
"""Register an activity function. Can be used as a decorator."""
|
||||
self._activities.append(activity_fn)
|
||||
return activity_fn
|
||||
|
||||
async def run(self) -> None:
|
||||
"""Start the worker. Blocks until shutdown."""
|
||||
temporal_client = (
|
||||
self._client.temporal
|
||||
if hasattr(self._client, "temporal")
|
||||
else self._client
|
||||
)
|
||||
self._worker = TemporalWorker(
|
||||
temporal_client,
|
||||
task_queue=self._queue,
|
||||
workflows=self._workflows,
|
||||
activities=self._activities,
|
||||
)
|
||||
await self._worker.run()
|
||||
|
||||
async def shutdown(self) -> None:
|
||||
"""Gracefully shutdown the worker."""
|
||||
if self._worker:
|
||||
await self._worker.shutdown()
|
||||
@@ -1,80 +0,0 @@
|
||||
"""Pre-built workflows for agent task orchestration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import timedelta
|
||||
|
||||
from temporalio import workflow
|
||||
from temporalio.common import RetryPolicy
|
||||
|
||||
|
||||
@dataclass
|
||||
class AgentTaskInput:
|
||||
"""Input for a single agent task."""
|
||||
|
||||
space_id: str
|
||||
agent_id: str
|
||||
task_title: str
|
||||
task_prompt: str
|
||||
timeout_seconds: int = 3600
|
||||
max_retries: int = 3
|
||||
|
||||
|
||||
@dataclass
|
||||
class AgentTaskOutput:
|
||||
"""Output from an agent task execution."""
|
||||
|
||||
result: str = ""
|
||||
error: str = ""
|
||||
elapsed_seconds: float = 0.0
|
||||
|
||||
|
||||
@workflow.defn
|
||||
class AgentTaskWorkflow:
|
||||
"""Execute a single agent task with retries and timeout."""
|
||||
|
||||
@workflow.run
|
||||
async def run(self, input: AgentTaskInput) -> AgentTaskOutput:
|
||||
return await workflow.execute_activity(
|
||||
"execute_agent_task",
|
||||
input,
|
||||
start_to_close_timeout=timedelta(seconds=input.timeout_seconds),
|
||||
retry_policy=RetryPolicy(maximum_attempts=input.max_retries),
|
||||
)
|
||||
|
||||
|
||||
@workflow.defn
|
||||
class PipelineWorkflow:
|
||||
"""Run agent tasks sequentially (pipeline)."""
|
||||
|
||||
@workflow.run
|
||||
async def run(self, tasks: list[AgentTaskInput]) -> list[AgentTaskOutput]:
|
||||
results: list[AgentTaskOutput] = []
|
||||
for task in tasks:
|
||||
result = await workflow.execute_activity(
|
||||
"execute_agent_task",
|
||||
task,
|
||||
start_to_close_timeout=timedelta(seconds=task.timeout_seconds),
|
||||
retry_policy=RetryPolicy(maximum_attempts=task.max_retries),
|
||||
)
|
||||
results.append(result)
|
||||
return results
|
||||
|
||||
|
||||
@workflow.defn
|
||||
class FanOutWorkflow:
|
||||
"""Run agent tasks in parallel (fan-out/fan-in)."""
|
||||
|
||||
@workflow.run
|
||||
async def run(self, tasks: list[AgentTaskInput]) -> list[AgentTaskOutput]:
|
||||
handles = []
|
||||
for task in tasks:
|
||||
handle = workflow.start_activity(
|
||||
"execute_agent_task",
|
||||
task,
|
||||
start_to_close_timeout=timedelta(seconds=task.timeout_seconds),
|
||||
retry_policy=RetryPolicy(maximum_attempts=task.max_retries),
|
||||
)
|
||||
handles.append(handle)
|
||||
return [await h for h in handles]
|
||||
@@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "hanzo-tasks"
|
||||
version = "0.1.0"
|
||||
description = "Hanzo Tasks SDK — Durable workflow execution for AI agents (powered by Temporal)"
|
||||
version = "0.2.0"
|
||||
description = "Hanzo Tasks client — start, signal, query and schedule durable workflows over /v1/tasks"
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
license = { text = "MIT" }
|
||||
@@ -19,9 +19,9 @@ classifiers = [
|
||||
"Topic :: Software Development :: Libraries :: Python Modules",
|
||||
"Typing :: Typed",
|
||||
]
|
||||
keywords = ["hanzo", "temporal", "tasks", "workflow", "durable", "agents"]
|
||||
keywords = ["hanzo", "tasks", "workflow", "durable", "agents"]
|
||||
dependencies = [
|
||||
"temporalio>=1.9.0",
|
||||
"httpx>=0.27",
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
@@ -32,7 +32,6 @@ Documentation = "https://hanzo.ai/docs/tasks"
|
||||
[project.optional-dependencies]
|
||||
dev = [
|
||||
"pytest>=7.0.0",
|
||||
"pytest-asyncio>=0.26.0",
|
||||
]
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
|
||||
@@ -1,317 +1,79 @@
|
||||
"""Hanzo Tasks test suite.
|
||||
|
||||
Unit tests for client config, worker registration, dataclasses,
|
||||
workflow decorators, and activity executor wiring. No Temporal
|
||||
server required.
|
||||
"""
|
||||
|
||||
import inspect
|
||||
from dataclasses import asdict
|
||||
import json
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
|
||||
from hanzo_tasks import (
|
||||
AgentTaskInput,
|
||||
AgentTaskOutput,
|
||||
AgentTaskWorkflow,
|
||||
Client,
|
||||
FanOutWorkflow,
|
||||
PipelineWorkflow,
|
||||
TasksConfig,
|
||||
Worker,
|
||||
WorkflowHandle,
|
||||
execute_agent_task,
|
||||
set_agent_executor,
|
||||
)
|
||||
|
||||
|
||||
# -- config ------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTasksConfig:
|
||||
def test_defaults(self):
|
||||
cfg = TasksConfig()
|
||||
assert cfg.address == "localhost:7233"
|
||||
assert cfg.namespace == "hanzo"
|
||||
assert cfg.tls is False
|
||||
|
||||
def test_custom(self):
|
||||
cfg = TasksConfig(address="temporal.prod:7233", namespace="prod", tls=True)
|
||||
assert cfg.address == "temporal.prod:7233"
|
||||
assert cfg.namespace == "prod"
|
||||
assert cfg.tls is True
|
||||
|
||||
|
||||
# -- dataclasses -------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAgentTaskInput:
|
||||
def test_defaults(self):
|
||||
inp = AgentTaskInput(
|
||||
space_id="sp-1",
|
||||
agent_id="ag-1",
|
||||
task_title="Fix bug",
|
||||
task_prompt="Fix the null pointer in main.go",
|
||||
)
|
||||
assert inp.space_id == "sp-1"
|
||||
assert inp.agent_id == "ag-1"
|
||||
assert inp.task_title == "Fix bug"
|
||||
assert inp.task_prompt == "Fix the null pointer in main.go"
|
||||
assert inp.timeout_seconds == 3600
|
||||
assert inp.max_retries == 3
|
||||
|
||||
def test_custom_timeout(self):
|
||||
inp = AgentTaskInput(
|
||||
space_id="sp-2",
|
||||
agent_id="ag-2",
|
||||
task_title="Deploy",
|
||||
task_prompt="Deploy to prod",
|
||||
timeout_seconds=600,
|
||||
max_retries=1,
|
||||
)
|
||||
assert inp.timeout_seconds == 600
|
||||
assert inp.max_retries == 1
|
||||
|
||||
def test_serializes_to_dict(self):
|
||||
inp = AgentTaskInput(
|
||||
space_id="sp-1",
|
||||
agent_id="ag-1",
|
||||
task_title="T",
|
||||
task_prompt="P",
|
||||
)
|
||||
d = asdict(inp)
|
||||
assert d["space_id"] == "sp-1"
|
||||
assert d["agent_id"] == "ag-1"
|
||||
assert d["task_title"] == "T"
|
||||
assert d["task_prompt"] == "P"
|
||||
assert d["timeout_seconds"] == 3600
|
||||
assert d["max_retries"] == 3
|
||||
assert len(d) == 6
|
||||
|
||||
|
||||
class TestAgentTaskOutput:
|
||||
def test_defaults(self):
|
||||
out = AgentTaskOutput()
|
||||
assert out.result == ""
|
||||
assert out.error == ""
|
||||
assert out.elapsed_seconds == 0.0
|
||||
|
||||
def test_success(self):
|
||||
out = AgentTaskOutput(result="done", elapsed_seconds=1.5)
|
||||
assert out.result == "done"
|
||||
assert out.error == ""
|
||||
assert out.elapsed_seconds == 1.5
|
||||
|
||||
def test_error(self):
|
||||
out = AgentTaskOutput(error="timeout", elapsed_seconds=3600.0)
|
||||
assert out.error == "timeout"
|
||||
assert out.result == ""
|
||||
|
||||
def test_serializes_to_dict(self):
|
||||
out = AgentTaskOutput(result="ok", elapsed_seconds=0.1)
|
||||
d = asdict(out)
|
||||
assert d == {"result": "ok", "error": "", "elapsed_seconds": 0.1}
|
||||
|
||||
|
||||
# -- workflow decorators -----------------------------------------------------
|
||||
|
||||
|
||||
class TestWorkflowDecorators:
|
||||
def test_agent_task_workflow_has_run(self):
|
||||
assert hasattr(AgentTaskWorkflow, "run")
|
||||
assert inspect.iscoroutinefunction(AgentTaskWorkflow.run)
|
||||
|
||||
def test_pipeline_workflow_has_run(self):
|
||||
assert hasattr(PipelineWorkflow, "run")
|
||||
assert inspect.iscoroutinefunction(PipelineWorkflow.run)
|
||||
|
||||
def test_fanout_workflow_has_run(self):
|
||||
assert hasattr(FanOutWorkflow, "run")
|
||||
assert inspect.iscoroutinefunction(FanOutWorkflow.run)
|
||||
|
||||
def test_workflow_classes_are_distinct(self):
|
||||
assert AgentTaskWorkflow is not PipelineWorkflow
|
||||
assert PipelineWorkflow is not FanOutWorkflow
|
||||
|
||||
|
||||
# -- worker ------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestWorker:
|
||||
def test_init_defaults(self):
|
||||
w = Worker(client=None, queue="test-q")
|
||||
assert w._queue == "test-q"
|
||||
assert w._workflows == []
|
||||
assert w._activities == []
|
||||
assert w._worker is None
|
||||
|
||||
def test_register_workflow(self):
|
||||
w = Worker(client=None)
|
||||
|
||||
class MyWorkflow:
|
||||
pass
|
||||
|
||||
result = w.register_workflow(MyWorkflow)
|
||||
assert result is MyWorkflow
|
||||
assert MyWorkflow in w._workflows
|
||||
|
||||
def test_register_activity(self):
|
||||
w = Worker(client=None)
|
||||
|
||||
async def my_activity(input):
|
||||
return "ok"
|
||||
|
||||
result = w.register_activity(my_activity)
|
||||
assert result is my_activity
|
||||
assert my_activity in w._activities
|
||||
|
||||
def test_register_multiple(self):
|
||||
w = Worker(client=None)
|
||||
for i in range(5):
|
||||
w.register_workflow(type(f"Wf{i}", (), {}))
|
||||
assert len(w._workflows) == 5
|
||||
|
||||
def test_init_with_preloaded(self):
|
||||
workflows = [AgentTaskWorkflow, PipelineWorkflow]
|
||||
activities = [execute_agent_task]
|
||||
w = Worker(client=None, workflows=workflows, activities=activities)
|
||||
assert len(w._workflows) == 2
|
||||
assert len(w._activities) == 1
|
||||
|
||||
def test_does_not_mutate_caller_list(self):
|
||||
workflows: list = []
|
||||
w = Worker(client=None, workflows=workflows)
|
||||
w.register_workflow(AgentTaskWorkflow)
|
||||
# The caller's original list should not be modified since we
|
||||
# pass a new list via `or []`, but if caller passes a list,
|
||||
# it IS the same reference. That's expected Python behavior.
|
||||
# Just verify worker has the workflow.
|
||||
assert AgentTaskWorkflow in w._workflows
|
||||
|
||||
|
||||
# -- executor wiring ---------------------------------------------------------
|
||||
|
||||
|
||||
class TestSetAgentExecutor:
|
||||
def test_set_and_reset(self):
|
||||
import hanzo_tasks.activities as act
|
||||
|
||||
original = act._agent_executor
|
||||
|
||||
async def my_exec(input):
|
||||
return "executed"
|
||||
|
||||
set_agent_executor(my_exec)
|
||||
assert act._agent_executor is my_exec
|
||||
|
||||
# Restore
|
||||
act._agent_executor = original
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_without_executor_raises(self):
|
||||
import hanzo_tasks.activities as act
|
||||
|
||||
saved = act._agent_executor
|
||||
act._agent_executor = None
|
||||
try:
|
||||
with pytest.raises(RuntimeError, match="No agent executor registered"):
|
||||
await execute_agent_task(None)
|
||||
finally:
|
||||
act._agent_executor = saved
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_execute_with_executor(self):
|
||||
import hanzo_tasks.activities as act
|
||||
|
||||
saved = act._agent_executor
|
||||
|
||||
async def mock_exec(input):
|
||||
return AgentTaskOutput(result=f"done:{input.task_title}", elapsed_seconds=0.01)
|
||||
|
||||
set_agent_executor(mock_exec)
|
||||
try:
|
||||
inp = AgentTaskInput(
|
||||
space_id="sp-1",
|
||||
agent_id="ag-1",
|
||||
task_title="test",
|
||||
task_prompt="do it",
|
||||
)
|
||||
out = await execute_agent_task(inp)
|
||||
assert out.result == "done:test"
|
||||
assert out.elapsed_seconds == 0.01
|
||||
finally:
|
||||
act._agent_executor = saved
|
||||
|
||||
|
||||
# -- workflow handle ---------------------------------------------------------
|
||||
|
||||
|
||||
class TestWorkflowHandle:
|
||||
def test_id(self):
|
||||
class FakeHandle:
|
||||
id = "wf-123"
|
||||
result_run_id = "run-456"
|
||||
|
||||
h = WorkflowHandle(FakeHandle())
|
||||
assert h.id == "wf-123"
|
||||
assert h.run_id == "run-456"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cancel(self):
|
||||
cancelled = False
|
||||
|
||||
class FakeHandle:
|
||||
id = "wf-1"
|
||||
result_run_id = "run-1"
|
||||
|
||||
async def cancel(self):
|
||||
nonlocal cancelled
|
||||
cancelled = True
|
||||
|
||||
h = WorkflowHandle(FakeHandle())
|
||||
await h.cancel()
|
||||
assert cancelled
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_signal(self):
|
||||
signals = []
|
||||
|
||||
class FakeHandle:
|
||||
id = "wf-1"
|
||||
result_run_id = "run-1"
|
||||
|
||||
async def signal(self, name, data=None):
|
||||
signals.append((name, data))
|
||||
|
||||
h = WorkflowHandle(FakeHandle())
|
||||
await h.signal("pause", {"reason": "lunch"})
|
||||
assert signals == [("pause", {"reason": "lunch"})]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_result(self):
|
||||
class FakeHandle:
|
||||
id = "wf-1"
|
||||
result_run_id = "run-1"
|
||||
|
||||
async def result(self, result_type=None):
|
||||
return "the-result"
|
||||
|
||||
h = WorkflowHandle(FakeHandle())
|
||||
assert await h.result() == "the-result"
|
||||
|
||||
|
||||
# -- __init__ exports --------------------------------------------------------
|
||||
|
||||
|
||||
class TestExports:
|
||||
def test_version(self):
|
||||
import hanzo_tasks
|
||||
|
||||
assert hanzo_tasks.__version__ == "0.1.0"
|
||||
|
||||
def test_all_exports_importable(self):
|
||||
import hanzo_tasks
|
||||
|
||||
for name in hanzo_tasks.__all__:
|
||||
assert hasattr(hanzo_tasks, name), f"{name} not found in hanzo_tasks"
|
||||
from hanzo_tasks import Client, TasksConfig, TasksError
|
||||
|
||||
|
||||
def make(handler):
|
||||
return Client(TasksConfig(token="tok", namespace="acme", host="https://api.example"), transport=httpx.MockTransport(handler))
|
||||
|
||||
|
||||
def test_start_posts_the_engine_shape():
|
||||
seen = {}
|
||||
|
||||
def handler(req: httpx.Request) -> httpx.Response:
|
||||
seen["method"], seen["path"], seen["auth"] = req.method, req.url.path, req.headers.get("authorization")
|
||||
seen["body"] = json.loads(req.content)
|
||||
return httpx.Response(200, json={"workflowId": seen["body"]["workflowId"], "runId": "r-1", "status": "Running"})
|
||||
|
||||
with make(handler) as c:
|
||||
wf = c.start("agent.review", task_queue="agents", input={"repo": "acme/widgets"}, workflow_id="wf-1")
|
||||
assert seen["method"] == "POST" and seen["path"] == "/v1/tasks/namespaces/acme/workflows"
|
||||
assert seen["auth"] == "Bearer tok"
|
||||
assert seen["body"]["workflowType"] == {"name": "agent.review"} and seen["body"]["taskQueue"] == {"name": "agents"}
|
||||
assert seen["body"]["input"] == {"repo": "acme/widgets"} and seen["body"]["requestId"]
|
||||
assert wf.workflow_id == "wf-1" and wf.run_id == "r-1" and wf.status == "Running"
|
||||
|
||||
|
||||
def test_signal_query_cancel_reset_routes():
|
||||
calls = []
|
||||
|
||||
def handler(req: httpx.Request) -> httpx.Response:
|
||||
calls.append((req.method, req.url.path, req.url.query.decode(), json.loads(req.content) if req.content else None))
|
||||
return httpx.Response(200, json={"workflowId": "wf-1", "runId": "r-1", "answer": 42})
|
||||
|
||||
with make(handler) as c:
|
||||
c.signal("wf-1", "approve", {"by": "z"}, run_id="r-1")
|
||||
out = c.query("wf-1", "status")
|
||||
c.cancel("wf-1", reason="done")
|
||||
c.terminate("wf-1", reason="hard")
|
||||
c.reset("wf-1", event_id=7, run_id="r-1", reason="replay")
|
||||
c.signal_with_start("agent.review", task_queue="agents", signal="kick", payload=1, workflow_id="wf-2")
|
||||
paths = [(m, p) for m, p, _, _ in calls]
|
||||
assert paths == [
|
||||
("POST", "/v1/tasks/namespaces/acme/workflows/wf-1/signal"),
|
||||
("POST", "/v1/tasks/namespaces/acme/workflows/wf-1/query"),
|
||||
("POST", "/v1/tasks/namespaces/acme/workflows/wf-1/cancel"),
|
||||
("POST", "/v1/tasks/namespaces/acme/workflows/wf-1/terminate"),
|
||||
("POST", "/v1/tasks/namespaces/acme/workflows/wf-1/reset"),
|
||||
("POST", "/v1/tasks/namespaces/acme/workflows/signal-with-start"),
|
||||
]
|
||||
assert calls[0][2] == "runId=r-1" and calls[0][3] == {"name": "approve", "payload": {"by": "z"}}
|
||||
assert calls[1][3] == {"queryType": "status", "args": None} and out["answer"] == 42
|
||||
assert calls[4][3] == {"runId": "r-1", "eventId": 7, "reason": "replay", "identity": ""}
|
||||
assert calls[5][3]["signalName"] == "kick" and calls[5][3]["signalPayload"] == 1
|
||||
|
||||
|
||||
def test_schedules_and_errors():
|
||||
calls = []
|
||||
|
||||
def handler(req: httpx.Request) -> httpx.Response:
|
||||
calls.append((req.method, req.url.path))
|
||||
if req.url.path.endswith("/schedules/nope"):
|
||||
return httpx.Response(404, json={"error": "schedule not found"})
|
||||
return httpx.Response(200, json={"ok": True})
|
||||
|
||||
with make(handler) as c:
|
||||
c.create_schedule({"id": "nightly", "spec": {"cron": "0 2 * * *"}, "action": {"workflowType": "agent.digest"}})
|
||||
c.trigger_schedule("nightly")
|
||||
c.delete_schedule("nightly")
|
||||
with pytest.raises(TasksError) as e:
|
||||
c.schedule("nope")
|
||||
assert e.value.status == 404 and "schedule not found" in str(e.value)
|
||||
assert calls[:3] == [
|
||||
("POST", "/v1/tasks/namespaces/acme/schedules"),
|
||||
("POST", "/v1/tasks/namespaces/acme/schedules/nightly/trigger"),
|
||||
("DELETE", "/v1/tasks/namespaces/acme/schedules/nightly"),
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user