Testing your agent

# Terminal 1 — server
awaithumans dev

# Terminal 2 — your agent
python my_agent.py
# (Agent blocks; open http://localhost:3001, approve, agent unblocks.)

# smoke.py — the shape of an automated end-to-end test
import json
import re
import os
import tempfile
import time
from pathlib import Path

import requests
from awaithumans import await_human_sync
from pydantic import BaseModel

EMAIL_DIR = Path(tempfile.mkdtemp(prefix="awaithumans-smoke-"))
SERVER = "http://localhost:3001"
TOKEN = os.environ["AWAITHUMANS_ADMIN_API_TOKEN"]

# 1. Create a file-transport email identity
requests.post(
    f"{SERVER}/api/channels/email/identities",
    headers={"Authorization": f"Bearer {TOKEN}"},
    json={
        "id": "smoke",
        "display_name": "Smoke",
        "from_email": "smoke@example.test",
        "transport": "file",
        "transport_config": {"dir": str(EMAIL_DIR)},
    },
)

# 2. Kick off await_human (in a thread so we can drive the response)
class WirePayload(BaseModel):
    vendor: str
    amount_usd: int


class Decision(BaseModel):
    approved: bool


import threading
result_box = {}
def run_agent():
    result_box["v"] = await_human_sync(
        task="Approve wire?",
        payload_schema=WirePayload,
        payload=WirePayload(vendor="Atlas Logistics", amount_usd=48200),
        response_schema=Decision,
        timeout_seconds=120,
        notify=["email+smoke:reviewer@example.test"],
    )
threading.Thread(target=run_agent, daemon=True).start()

# 3. Poll the file dir for the rendered email
deadline = time.time() + 30
while time.time() < deadline:
    files = list(EMAIL_DIR.glob("*.json"))
    if files:
        break
    time.sleep(0.5)

email = json.loads(files[0].read_text())
approve_url = re.search(r'href="([^"]+/email/action/[^"]+)"', email["html"]).group(1)

# 4. POST to the magic-link URL — same as a human clicking "Approve"
requests.post(approve_url)

# 5. Agent unblocks; assert
time.sleep(1)
assert result_box["v"].approved is True

# In a fixture / test setup, after kicking off your agent:
import requests, os

# Get the most recent task ID (or pre-stamp `idempotency_key` so you know it)
tasks = requests.get(
    f"http://localhost:3001/api/tasks?limit=1",
    headers={"Authorization": f"Bearer {os.environ['AWAITHUMANS_ADMIN_API_TOKEN']}"},
).json()
task_id = tasks["items"][0]["id"]

# Complete it
requests.post(
    f"http://localhost:3001/api/tasks/{task_id}/complete",
    headers={"Authorization": f"Bearer {os.environ['AWAITHUMANS_ADMIN_API_TOKEN']}"},
    json={"response": {"approved": True, "notes": "test fixture"}},
)
# Your agent's await_human() resolves on next poll cycle (~25s max).

from pydantic import BaseModel
from awaithumans import await_human_sync


class RefundRequest(BaseModel):
    order_id: str
    amount_usd: int


class RefundDecision(BaseModel):
    approved: bool


def test_refund_approval():
    decision = await_human_sync(
        task="Approve refund for test order",
        payload_schema=RefundRequest,
        payload=RefundRequest(order_id="test-A-4721", amount_usd=250),
        response_schema=RefundDecision,
        timeout_seconds=60,
        idempotency_key="test:refund-approval",   # ← stable per-test
    )
    assert decision.approved is True

idempotency_key=f"test:refund-approval:{uuid.uuid4()}"