openphn 0.3.0

Python SDK for OpenPhn — agentic voice calls with structured outcome JSON

OpenPhn — Python SDK

Agentic voice calls with structured JSON outcomes.

pip install openphn

Quickstart

from openphn import OpenPhn

with OpenPhn(api_key="sk_live_...") as client:
    result = client.call(
        to="[phone]",
        objective="Confirm order A-14421 ships today",
        outcome_schema={
            "will_ship_today": {"type": "boolean"},
            "tracking":        {"type": "string",   "optional": True},
            "eta":             {"type": "datetime", "optional": True},
        },
        consent_type="existing_business_relationship",
    )
    print(result["outcome"])
    # {'will_ship_today': True, 'tracking': '1Z...', 'eta': '2026-04-23T18:00:00-07:00'}

client.call() blocks until the call finishes by default. Pass wait=False to return as soon as the server accepts the request, then pair with a webhook (see Webhooks).

Why OpenPhn

• Structured JSON per call — you define outcome_schema; we return typed fields with per-field confidence scores and ambiguity flags.
• Compliance at the API boundary — consent_type enum required; call-hour rules (8am–9pm in destination's timezone) enforced server-side; DNC scrubbing native.
• Hosted MCP server — any MCP-aware client (Claude Desktop, Cursor) can place calls via this platform as a tool.
• Published-and-frozen flows — edit your graph freely; publish freezes the artifact so runtime behavior doesn't silently drift.

Conceptual docs: docs.openphn.com. Interactive API reference: api.openphn.com/docs.

Async

import asyncio
from openphn import AsyncOpenPhn

async def main():
    async with AsyncOpenPhn(api_key="sk_live_...") as client:
        r = await client.call(
            to="[phone]",
            objective="Follow up on the Thursday appointment",
            outcome_schema={"attended": {"type": "boolean"}},
            consent_type="existing_business_relationship",
        )
        print(r["outcome"])

asyncio.run(main())

Typed errors

Every non-2xx becomes a typed subclass of OpenPhnError.

from openphn import (
    OpenPhn,
    DNCBlockedError,
    RateLimitError,
    ValidationError,
    VerificationPendingError,
)

try:
    client.call(to="[phone]", ...)
except DNCBlockedError:
    # Destination on suppression list. Do not retry.
    pass
except VerificationPendingError:
    # Account still `pending_review`. Wait for admin approval.
    pass
except RateLimitError as e:
    time.sleep(e.retry_after_seconds or 30)
    client.call(...)
except ValidationError as e:
    print(e.error_code, e.message)

Exception	Status	When
AuthenticationError	401	Missing / malformed / revoked key
PermissionError	403	Valid key, insufficient scope or number_id pinning
DNCBlockedError	403 (dnc_blocked)	Destination on suppression list
VerificationPendingError	403 (verification_pending)	Account not yet approved
NotFoundError	404	Resource doesn't exist for your tenant
ConflictError	409	idempotency_key reused with different body
ValidationError	422	Semantically-invalid request
ConsentError	422 (consent_*)	Consent type missing / invalid
RateLimitError	429	Over per-key quota. Has .retry_after_seconds.
ServerError	5xx	Transient. Safe to retry idempotent requests.

Retries

Auto-retry on 429 + 5xx for idempotent requests (GET, DELETE, and any POST with idempotency_key set). Exponential backoff with jitter; honors Retry-After. Override with max_retries=N (default 3, 0 disables).

Idempotency

call() and create_batch() accept idempotency_key. Re-sending with the same key within 24h returns the original response instead of creating a duplicate.

import uuid
key = f"order-{order_id}-{uuid.uuid4()}"
r1 = client.call(to="[phone]", ..., idempotency_key=key)
# network blip → safe retry
r2 = client.call(to="[phone]", ..., idempotency_key=key)
assert r1["call_id"] == r2["call_id"]

Webhooks

wh = client.create_webhook(
    url="https://example.com/webhooks/openphn",
    description="Prod call-completion handler",
)
print(wh["secret"])  # SHOWN ONCE — save it for HMAC verification

Verify HMAC-SHA256 on delivery:

import hmac, hashlib, time

def verify(raw_body: bytes, timestamp: str, signature: str, secret: str) -> None:
    expected = "sha256=" + hmac.new(
        secret.encode(),
        f"{timestamp}.{raw_body.decode()}".encode(),
        hashlib.sha256,
    ).hexdigest()
    if not hmac.compare_digest(expected, signature):
        raise ValueError("bad signature")
    if abs(time.time() - int(timestamp)) > 300:
        raise ValueError("timestamp too old")

Inspect delivery history (retries + response bodies) and manually replay:

for d in client.list_webhook_deliveries("wh_01HV..."):
    print(d["attempt"], d["status_code"], d["latency_ms"])

client.retry_webhook_delivery("wh_01HV...", "dlv_01HV...")

DNC (Do Not Call)

Upload up to 10,000 phone numbers per CSV request. E.164 normalized, NANP-only, deduped.

result = client.upload_dnc("suppressions.csv")
print(result["added_count"])

Subsequent call() to any listed number raises DNCBlockedError. Globally- suppressed numbers (federal DNC imports, platform bans) enforce the same way.

Iterate calls (auto-paged)

for call in client.iter_calls(status="delivered"):
    print(call["id"], call["outcome"])

# async
async for call in client.iter_calls(status="delivered"):
    ...

BYO Twilio (outbound)

verified = client.verify_twilio(
    account_sid="ACxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
    auth_token="your-twilio-auth-token",
)
print([n["phone_number"] for n in verified["numbers"]])

client.select_number(e164="[phone]")

Configure an inbound number

client.update_number(
    number_id="num_01HV...",
    greeting_text="Hi, this is the virtual receptionist for Northside Dental...",
    transfer_destinations=[
        {"label": "Front desk",      "phone": "[phone]", "enabled": True, "order": 0},
        {"label": "On-call manager", "phone": "[phone]", "enabled": True, "order": 1},
    ],
    recording_enabled=True,
)

Analytics summary

s = client.analytics_summary(range="30d")
print(f"{s['total_calls']} calls · {s['success_rate']:.1%} success · "
      f"greeting p50 {s['greeting_latency_p50_ms']}ms")

Whoami

me = client.me()
print(me["email"], me["verification_status"], me.get("scopes"))

Staging / self-hosted base URL

OpenPhn(api_key="sk_live_...", base_url="https://api.staging.openphn.com")

Compatibility

• Python 3.10+
• httpx 0.27+
• typing_extensions.NotRequired auto-used on 3.10; native typing.NotRequired on 3.11+

License

MIT.

Links

• Docs: docs.openphn.com
• REST reference: api.openphn.com/docs
• Source: github.com/knsgill/openphn
• Changelog: docs.openphn.com/docs/changelog