aithericon-hpi 0.1.0

Python SDK for the Aithericon HPI API — typed client for human-in-the-loop tasks

aithericon-hpi

Python SDK for the Aithericon HPI API — typed client for human-in-the-loop tasks.

Install

pip install aithericon-hpi

Quick Start

from hpi_client import HPIClient

client = HPIClient("https://app.hpi.dev", "htk_your-token")

# Create a task
task = client.create_task(
    title="Review deployment plan",
    steps=[{
        "id": "review",
        "title": "Review",
        "blocks": [
            {"type": "mdsvex", "content": "Please review the deployment plan for **v2.1**."},
            {"type": "input", "field": {"name": "approved", "label": "Approve?", "kind": "checkbox"}},
        ],
    }],
)

print(f"Task created: {task['task_id']}")
print(f"Magic link: {task['url']}")

# Wait for the human to complete it
result = client.wait_for_task(task["task_id"])
print(result["status"])  # "completed"
print(result["data"])    # {"approved": True}

API

Tasks

# Create
task = client.create_task(title="...", steps=[...])
task = client.create_task(
    title="...",
    steps=[...],
    assignee_email="[e-mail]",
    metadata={"env": "production"},
    sinks=[{"type": "webhook", "url": "https://..."}],
)

# Read
task = client.get_task("task-id")
task = client.get_task("task-id", include_definition=True)
result = client.list_tasks(status="pending", sort="newest", limit=10)

# Lifecycle
client.complete_task("task-id", data={"approved": True})
client.cancel_task("task-id", reason="No longer needed")
client.notify_task("task-id")

# Wait (long-poll)
result = client.wait_for_task("task-id", timeout=300)
if result.get("timed_out"):
    print("Still pending")

# Templates
templates = client.list_task_templates()

Task Links

link = client.create_task_link("task-id", assignee_email="[e-mail]")
links = client.list_task_links("task-id")
client.delete_task_link("task-id", "link-id")

Processes

proc = client.create_process(
    namespace="deployments",
    name="Deploy v2.1",
    steps=[
        {"key": "build", "label": "Build"},
        {"key": "review", "label": "Human Review", "human": True},
        {"key": "deploy", "label": "Deploy"},
    ],
)

client.update_process(proc["process_id"], {"type": "step_started", "step": "build"})
client.update_process(proc["process_id"], {"type": "step_completed", "step": "build"})
client.update_process(proc["process_id"], {"type": "completed"})

proc = client.get_process(proc["process_id"])
result = client.list_processes(status="running", namespace="deployments")

Process Links

link = client.create_process_link("process-id", assignee_email="[e-mail]")
links = client.list_process_links("process-id")

File Upload

with open("report.pdf", "rb") as f:
    result = client.upload_file(f.read(), "report.pdf", "application/pdf")
print(result["url"])

Error Handling

from hpi_client.client import HPIError

try:
    client.get_task("nonexistent")
except HPIError as e:
    print(e.status)   # 404
    print(e.error)    # "Task not found"
    print(e.details)  # None or list of validation details

Context Manager

with HPIClient("https://app.hpi.dev", "htk_your-token") as client:
    task = client.create_task(title="...", steps=[...])
# connection closed automatically

Self-Hosted

client = HPIClient("https://hpi.yourcompany.com", "htk_your-token")

License

Apache License 2.0