mirror of
https://github.com/hanzoai/agents.git
synced 2026-10-30 18:48:36 +00:00
zeekayandGitHub
5b5cf9e9db
Transforms hanzoai/agents from a marketplace config repo into a full agent orchestration platform by merging the AgentField infrastructure: - Go control plane (K8s-like agent orchestration, service mesh, DAG workflows) - Python SDK (FastAPI-based agent framework, LiteLLM, memory, DID/VC) - TypeScript SDK (@hanzo/agents - Vercel AI SDK, MCP, workflows) - Go SDK (idiomatic agent builder + client) - React 19 embedded dashboard (XyFlow DAGs, Recharts, Radix UI) - Docker/K8s/Helm deployment configs - Examples: hello world, deep research, RAG, simulation, benchmarks - Functional test suite across Python/Go/TypeScript All references rebranded: AgentField → Hanzo Agents - Module: github.com/hanzoai/agents/control-plane - Python: hanzo_agents - TypeScript: @hanzo/agents - Go: github.com/hanzoai/agents/sdk/go - Env vars: HANZO_AGENTS_* - Config: hanzo-agents.yaml Previous marketplace configs moved to config/ directory.
71 lines
1.6 KiB
Python
71 lines
1.6 KiB
Python
"""Canonical execution status utilities for the Hanzo Agents SDK."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Optional, Set, Tuple
|
|
|
|
CANONICAL_STATUSES: Tuple[str, ...] = (
|
|
"pending",
|
|
"queued",
|
|
"running",
|
|
"succeeded",
|
|
"failed",
|
|
"cancelled",
|
|
"timeout",
|
|
"unknown",
|
|
)
|
|
|
|
CANONICAL_STATUS_SET: Set[str] = set(CANONICAL_STATUSES)
|
|
|
|
_STATUS_ALIASES = {
|
|
"success": "succeeded",
|
|
"successful": "succeeded",
|
|
"completed": "succeeded",
|
|
"complete": "succeeded",
|
|
"done": "succeeded",
|
|
"ok": "succeeded",
|
|
"error": "failed",
|
|
"failure": "failed",
|
|
"errored": "failed",
|
|
"canceled": "cancelled",
|
|
"cancel": "cancelled",
|
|
"timed_out": "timeout",
|
|
"wait": "queued",
|
|
"waiting": "queued",
|
|
"in_progress": "running",
|
|
"processing": "running",
|
|
}
|
|
|
|
TERMINAL_STATUSES: Set[str] = {"succeeded", "failed", "cancelled", "timeout"}
|
|
|
|
|
|
def normalize_status(status: Optional[str]) -> str:
|
|
"""Return the canonical representation of a status string."""
|
|
|
|
if status is None:
|
|
return "unknown"
|
|
|
|
normalized = status.strip().lower()
|
|
if not normalized:
|
|
return "unknown"
|
|
|
|
if normalized in CANONICAL_STATUS_SET:
|
|
return normalized
|
|
|
|
return _STATUS_ALIASES.get(normalized, "unknown")
|
|
|
|
|
|
def is_terminal(status: Optional[str]) -> bool:
|
|
"""Return True if the provided status represents a terminal state."""
|
|
|
|
return normalize_status(status) in TERMINAL_STATUSES
|
|
|
|
|
|
__all__ = [
|
|
"CANONICAL_STATUSES",
|
|
"CANONICAL_STATUS_SET",
|
|
"TERMINAL_STATUSES",
|
|
"normalize_status",
|
|
"is_terminal",
|
|
]
|