Python SDK - Await Humans

Install

Pick one of two starting points based on what you’re doing:

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

Most people want the first. Use the bare SDK install only when your agent runs in a lambda, edge runtime, or container where you just need an HTTP client talking to a server hosted elsewhere — bare install is httpx + pydantic and nothing else.

Adapters and verifiers compose with either base

Extras stack. Comma-separate them in one bracket to install in a single command:

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

All extras

Extra	Adds
`server`	FastAPI server, CLI, bundled dashboard
`temporal`	Temporal workflow adapter
`langgraph`	LangGraph interrupt/resume adapter
`verifier-claude`	Anthropic SDK (for Claude verification)
`verifier-openai`	OpenAI SDK
`verifier-gemini`	Google Generative AI SDK
`verifier-azure`	OpenAI SDK (used against Azure endpoints)

Python 3.10+ required.

await_human / await_human_sync

The primitive. Async and sync flavors:

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())

Parameters

Parameter	Type	Required	Description
`task`	`str`	yes	Human-readable task description (shown in the dashboard / Slack / email).
`payload_schema`	`type[BaseModel]`	yes	Pydantic class that defines the data the human reviews.
`payload`	`BaseModel`	yes	Instance of `payload_schema` — the actual data.
`response_schema`	`type[BaseModel]`	yes	Pydantic class for the response form. Drives the UI.
`timeout_seconds`	`int`	yes	Min 60 (1 min), max 2,592,000 (30 days).
`assign_to`	`str \| list[str] \| dict \| None`	no	Routing target. See Routing.
`notify`	`list[str] \| None`	no	Channel destinations (`slack:#x`, `email:y@z`, …).
`verifier`	`VerifierConfig \| None`	no	Server-side LLM verification. See Verifier.
`idempotency_key`	`str \| None`	no	Default: `sha256(task, payload)[:32]`. See Idempotency.
`redact_payload`	`bool`	no	Default `False`. When true, audit log + verifier prompt are scrubbed.
`server_url`	`str \| None`	no	Override `AWAITHUMANS_URL` env var / discovery file.
`api_key`	`str \| None`	no	Override `AWAITHUMANS_ADMIN_API_TOKEN` env var / discovery file.

Returns

An instance of response_schema — Pydantic-validated. If validation fails, raises SchemaValidationError.

Raises

Exception	When
`TaskTimeoutError`	Task hit `timeout_seconds` without completion.
`TaskCancelledError`	Agent or operator cancelled.
`VerificationExhaustedError`	Verifier rejected `max_attempts` times.
`SchemaValidationError`	Response didn’t match `response_schema`.
`TaskNotFoundError`	Task ID disappeared (DB reset?).
`TaskCreateError`	Server rejected `POST /api/tasks`.
`PollError`	Long-poll returned non-200.
`ServerUnreachableError`	Couldn’t connect to `server_url`.
`MarketplaceNotAvailableError`	Used `assign_to={"marketplace": True}`. Reserved for Phase 3.

All inherit from AwaitHumansError.

Verifier helpers

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
)

Each returns a VerifierConfig you pass to await_human(verifier=...). The actual LLM call runs server-side. See Verifier.

Adapters

Temporal

# 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

See Temporal.

LangGraph

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

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

See LangGraph.

CLI

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

awaithumans dev auto-generates PAYLOAD_KEY, ADMIN_API_TOKEN, and the SQLite DB if missing. Writes ~/.awaithumans-dev.json so the SDK auto-discovers the running server.

Configuration

The SDK reads configuration in this order (first match wins):

Function-call args (server_url=, api_key=)
Environment variables (AWAITHUMANS_URL, AWAITHUMANS_ADMIN_API_TOKEN)
Discovery file (~/.awaithumans-dev.json)
Defaults (http://localhost:3001, no token)

Production deployments should use environment variables. The discovery file is a dev affordance.

Type stubs

The package ships with py.typed. mypy/pyright pick up the types automatically:

# (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.

​Install

​Adapters and verifiers compose with either base

​All extras

​await_human / await_human_sync

​Parameters

​Returns

​Raises

​Verifier helpers

​Adapters

​Temporal

​LangGraph

​CLI

​Configuration

​Type stubs