Areteos Python SDK

Python client for the Areteos Workflow Runtime API. Define workflows, trigger runs, and monitor execution with full type safety.

Installation

pip install areteos

Quick Start

from areteos import AreteosClient, Step, Edge, InputMapping

client = AreteosClient(
    base_url="https://areteos.example.com",
    api_key="areteos_your_key",
    engagement_id="your-engagement-uuid",
)

# Register a workflow
workflow = client.create_workflow(
    "Document Pipeline",
    steps=[
        Step.python("extract", "Extract Text", code="""
import json, sys
data = json.load(open(sys.argv[1])) if len(sys.argv) > 1 else {}
print(json.dumps({"text": data.get("raw_text", "")}))
"""),
        Step.agent(
            "summarize", "Summarize",
            system_prompt="Summarize the document concisely.",
            input_mappings=[
                InputMapping(source_step_id="extract", source_field="text", target_field="document")
            ],
        ),
    ],
    edges=[Edge(from_step_id="extract", to_step_id="summarize")],
)

# Trigger and wait
run = client.trigger(workflow.current_version_id, input={"raw_text": "..."})
result = client.wait_for_run(run.id, timeout=300)
print(result.output)

Async Usage

from areteos import AreteosAsyncClient, Step

async with AreteosAsyncClient(
    base_url="https://areteos.example.com",
    api_key="areteos_your_key",
    engagement_id="your-engagement-uuid",
) as client:
    workflow = await client.create_workflow("My Workflow", steps=[...])
    run = await client.trigger(workflow.current_version_id)
    result = await client.wait_for_run(run.id)

Step Types

Agent (LLM)

Step.agent(
    "analyze", "Analyze",
    model="capable",           # fast, capable, thinking
    strategy="react",          # react, adaptive, cot, cod, tot, got, trm
    system_prompt="You are an analyst.",
    max_turns=10,
)

Python (Inline Code)

Step.python(
    "transform", "Transform",
    code="import json, sys\ndata = json.load(open(sys.argv[1]))\nprint(json.dumps({'count': len(data)}))",
    timeout_ms=30000,
)

Human-in-the-Loop

Step.hitl(
    "review", "Manager Review",
    request_type="approval",   # approval, input, review
    timeout_minutes=1440,      # 24 hours
)

Data Flow

Connect steps with edges and input mappings:

steps = [
    Step.python("extract", "Extract", code="..."),
    Step.agent(
        "summarize", "Summarize",
        input_mappings=[
            InputMapping(source_step_id="extract", source_field="text", target_field="document"),
            InputMapping(source_step_id=None, source_field="config", target_field="settings"),  # from run input
        ],
    ),
]
edges = [Edge(from_step_id="extract", to_step_id="summarize")]

Error Handling

from areteos import AreteosError, ApiError, ForbiddenError, NotFoundError, ValidationError, TimeoutError

try:
    run = client.trigger(version_id)
    result = client.wait_for_run(run.id, timeout=60)
except ValidationError as e:
    print(f"Invalid workflow: {e}")
except ForbiddenError:
    print("Access denied — check your API key scope")
except TimeoutError:
    print("Run did not complete in time")
except ApiError as e:
    print(f"API error {e.status_code}: {e}")

Auto-Approve HITL

For CI/CD pipelines, auto-approve all human-in-the-loop requests:

result = client.wait_for_run(run.id, auto_approve=True)

Development

cd sdks/python
uv sync
uv run pytest
uv run ruff check .
uv run mypy src/