Python SDK

pip install "awaithumans[server]"   # to RUN the server + dashboard locally
pip install awaithumans              # to CALL a server (slim SDK, 2 deps)

pip install "awaithumans[server,temporal]"            # server + Temporal
pip install "awaithumans[server,verifier-claude]"     # server + Claude verifier
pip install "awaithumans[temporal,verifier-claude]"   # SDK + Temporal + Claude
pip install "awaithumans[server,temporal,verifier-claude]"  # all three

import asyncio
from pydantic import BaseModel
from awaithumans import await_human, await_human_sync


# Define the shape of the data your agent sends to the human...
class RefundPayload(BaseModel):
    amount: int
    customer: str


# ...and the shape of the response form they fill in.
class RefundDecision(BaseModel):
    approved: bool
    reason: str = ""


# Sync (from a regular Python script, Flask, Celery, etc.)
decision = await_human_sync(
    task="Approve refund?",
    payload_schema=RefundPayload,
    payload=RefundPayload(amount=250, customer="cus_123"),
    response_schema=RefundDecision,
    timeout_seconds=900,
)
print(decision.approved, decision.reason)


# Async (inside an existing asyncio context — agent frameworks, ASGI)
async def main() -> None:
    decision = await await_human(
        task="Approve refund?",
        payload_schema=RefundPayload,
        payload=RefundPayload(amount=250, customer="cus_123"),
        response_schema=RefundDecision,
        timeout_seconds=900,
    )
    print(decision.approved, decision.reason)


asyncio.run(main())

from awaithumans.verifiers.claude import claude_verifier
from awaithumans.verifiers.openai import openai_verifier
from awaithumans.verifiers.gemini import gemini_verifier
from awaithumans.verifiers.azure_openai import azure_openai_verifier

verifier = claude_verifier(
    instructions="Reject any refund decision that lacks a clear reason.",
    model="claude-sonnet-4-5",         # optional, has default
    max_attempts=3,                    # default 3
    api_key_env="ANTHROPIC_API_KEY",   # default
)

# Inside a Temporal workflow:
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
    from awaithumans.adapters.temporal import await_human

# In your web server (separate process):
from awaithumans.adapters.temporal import dispatch_signal

# Inside a LangGraph node:
from awaithumans.adapters.langgraph import await_human

# Driver — run the graph:
from awaithumans.adapters.langgraph import drive_human_loop

awaithumans dev               # boot server + auto-config for local dev
awaithumans add-user          # add a directory user (interactive)
awaithumans list-users        # list directory users
awaithumans version           # print version

# (using the RefundPayload / RefundDecision models defined earlier)
decision: RefundDecision = await_human_sync(
    task="Approve refund?",
    payload_schema=RefundPayload,
    payload=RefundPayload(amount=250, customer="cus_123"),
    response_schema=RefundDecision,
    timeout_seconds=900,
)
# `decision` is typed as RefundDecision — IDE autocomplete + mypy enforcement.