butt-dial-sdk 0.4.1

Client-side tunnel SDK for the Butt-Dial communication service — outbound, inbound callbacks, retry orchestration, and end-to-end diagnostic.

butt-dial-sdk

A stateless client-side tunnel to the Butt-Dial communication service for Python agentic systems. Ships outbound messaging, inbound callbacks, retry orchestration, an admin router, and a staged diagnostic with actionable remediation.

Status: 0.4.1. In production via iv-bknd. Sprint-9 onboarding handshake (branded email + webhook callback) shipped.

---

Install

pip install butt-dial-sdk                      # core: Agent, AccountsClient, InboundHandler, RetryWorker, doctor CLI
pip install 'butt-dial-sdk[router]'            # + FastAPI admin router

Python 3.11+. 218/218 tests green.

Building a host app? See AGENTS.md for the agent-readable contract — patterns to use, anti-patterns to avoid, and what to do on every breaking change.

Identity model — host owns the agent UUID

Butt-Dial 0.3.0 changed who generates agent_id. The host application generates the UUID for each agent (Maya, Sara, Yossi…) and tells BD at registration. BD echoes it back along with a minted per-agent bearer token. That keeps your host as the source of truth for agent identity; BD just holds the credentials and the channels.

Three tokens, three jobs:

Token	Held by	Used for
team_token	host backend	AccountsClient — provisioning, registration, billing, lifecycle
agent_token	per-agent runtime	Agent — sending and receiving on this agent's number
agent_id	host-supplied UUID	identifies the agent across both surfaces

Quickstart — register an agent and send a message

import uuid
from buttdial import AccountsClient, Agent

# 1. Workspace admin: register an agent (host owns the UUID).
accounts = AccountsClient("https://call.95percent.ai", team_token=TEAM_TOKEN)
agent_id = str(uuid.uuid4())  # host generates this
out = await accounts.register_agent(agent_id=agent_id, display_name="Maya")
agent_token = out["agent_token"]  # BD minted; persist to your vault

# 2. Per-agent runtime: send a message.
maya = Agent(
    base_url="https://call.95percent.ai",
    agent_id=agent_id,
    agent_token=agent_token,
)
result = await maya.send_message(to="[phone]", body="Hello from Maya")
print(result.message_sid)  # "SM..." on success

Or, for single-agent setups, pull all three from the environment:

export BUTT_DIAL_BASE_URL=https://call.95percent.ai
export BUTT_DIAL_AGENT_ID=<the UUID your host generated>
export BUTT_DIAL_AGENT_TOKEN=<token BD minted at registration>

from buttdial import Agent
maya = Agent.from_env()

See the examples/ folder for full working code.

---

What's included

The SDK bundles everything an agentic system needs to integrate with Butt-Dial reliably:

Component	What it does
Agent	Per-agent runtime. send_message() plus 13 typed WhatsApp action helpers (react, edit, forward, delete, typing, mark_read, send_poll, send_buttons, send_location, send_contact, post_status, set_chat_ttl, send_view_once) wrapping BD's comms_whatsapp_action. MCP/SSE transport with 3-attempt exponential-backoff retry. Provider mismatches raise ProviderCapabilityError.
InboundHandler	Decorator-based callbacks (@on_message, @on_delivery_receipt, @on_status_update). Ships WhatsApp payload parser + signature verification + subscription-challenge verification. Handler errors isolated.
RetryWorker	Background retry loop against a host-provided FailedMessageRepository protocol. Pluggable 2^n-minute backoff. Dead-letter transition with an optional on_dead_letter hook.
make_router()	FastAPI admin router: 7 endpoints for operators (overview, agents list, activate/deactivate, failed-message list/retry/dismiss). Plugs in via AgentDirectory + AdminFailedMessageRepository protocols.
run_diagnostic() / buttdial doctor	7-stage end-to-end health check with per-stage remediation messages. Runs as CLI or pytest fixture.
FakeButtDialServer	Programmable in-process fake for integration tests. Monkeypatches MCP at import boundary; no external process.
AccountsClient	Workspace admin REST wrapper. Onboarding + owner-token lifecycle (register → verify-email → set_phone → OTP → confirm_phone → rotate_token) plus per-agent provisioning (register_agent accepts a host-supplied agent_id, BD echoes it back with the minted agent_token). Typed exceptions per server error code. 0.4.0: register() accepts optional branding, return_url, webhook_url, verification_email for branded onboarding handshake.
buttdial.webhooks.verify_webhook_signature	Stripe-style HMAC-SHA256 verifier for the account.verified webhook BD posts after a user clicks their verification link. 5-minute replay protection, constant-time compare. (0.4.0+)

---

Onboard a tenant — the canonical pattern

For host apps that auto-register a Butt-Dial workspace when a customer signs up: pass branding / return_url / webhook_url to register(). The verification email is branded as your product, the verify page redirects users back to you, and BD posts back to your webhook the moment the user clicks. No polling, no "did it work?" dead-end.

from buttdial import AccountsClient
from buttdial.webhooks import verify_webhook_signature

accounts = AccountsClient("https://call.95percent.ai")

# 1. Register the workspace (auto-fired from your /signup handler).
resp = await accounts.register(
    org_name="Acme Inc",
    owner_email="[e-mail]",
    branding={
        "logoUrl":      "https://acme.com/logo-mark.svg",
        "primaryColor": "#00C9A7",
        "productName":  "Acme",
        "supportEmail": "[e-mail]",
    },
    return_url="https://acme.com/welcome",
    webhook_url="https://acme.com/api/webhooks/butt-dial/account-verified",
    # Optional: caller-supplied HTML email template.
    # Variables: {{verifyUrl}} (required), {{orgName}}, {{ownerEmail}}, {{expiresInMinutes}}.
    verification_email={
        "subject": "Verify your Acme workspace",
        "html":    "<p>Click <a href=\"{{verifyUrl}}\">here</a> to activate.</p>",
    },
)
# Capture once — webhookSecret is NOT retrievable later.
vault.put(workspace_id, "bd_webhook_secret", resp["webhookSecret"])
vault.put(workspace_id, "bd_poll_token",     resp["pollToken"])

# 2. Receive the webhook when the user clicks.
@app.post("/api/webhooks/butt-dial/account-verified")
async def bd_webhook(request):
    raw = await request.body()
    ok = verify_webhook_signature(
        raw,
        timestamp_header=request.headers.get("X-BD-Timestamp"),
        signature_header=request.headers.get("X-BD-Signature"),
        secret=vault.get(workspace_id, "bd_webhook_secret"),
    )
    if not ok:
        return JSONResponse({"error": "invalid signature"}, status_code=401)

    body = json.loads(raw)
    # body: {event, orgId, ownerEmail, teamToken, scopes, verifiedAt}
    workspace.bd_team_token = body["teamToken"]
    workspace.provisioned   = True
    workspace.save()
    return {"ok": True}

# 3. Phone confirmation, token rotation (workspace already verified at this point).
authed = accounts.with_token(workspace.bd_team_token)
challenge = await authed.set_phone("[phone]")
await authed.confirm_phone(challenge["challenge_id"], otp_from_user)
# Later:
new = await authed.rotate_token(otp_prompt=lambda hint:
    input(f"Enter code sent to {hint}: "))
vault.put(workspace_id, "bd_team_token", new["token"])

Idempotency: dedupe webhook deliveries by the X-BD-Delivery-Id header. BD retries (1s / 5s / 25s / 2m / 10m) until you 2xx, then dead-letters.

No webhook receiver? Drop webhook_url and the verify page redirects to <return_url>#bd_token=…&org_id=…&verified_at=… instead. Fragment never reaches a server, consumed by your frontend, which POSTs the team_token to your bknd. Webhook is preferred — works even if the user closes the tab.

Smoke-test only / dev? The bare register(org, email) form still works but emits a DeprecationWarning on every call: it produces a BD-branded email and gives no completion signal. Don't ship a host app with the bare form. See examples/with-onboarding-handshake/ for the working pattern.

---

How it fits together

The SDK is a stateless tunnel. It owns the Butt-Dial protocol and retry orchestration. The host app owns all persistence and identity via two protocols:

from buttdial import (
    Agent, InboundHandler, RetryWorker,
    FailedMessageRepository,    # protocol — host implements
    AgentDirectory,             # protocol — host implements
    make_router,
)

# 1. Build a per-agent runtime (one Agent per agent_id).
bd = Agent.from_env()

# 2. Outbound — call anywhere.
result = await bd.send_message(to="+1...", body="hi")

# 3. Inbound — register handlers, mount the router.
inbound = InboundHandler(whatsapp_verify_token="...", whatsapp_webhook_secret="...")

@inbound.on_message(channel="whatsapp")
async def on_msg(msg):
    await process(msg.sender, msg.text)

app.include_router(inbound.router, prefix="/api/webhook")

# 4. Retry — implement the 4 repo methods, start the worker.
class MyRepo:
    async def fetch_due(self, limit): ...
    async def mark_delivered(self, msg_id, message_sid, retry_count): ...
    async def increment_retry(self, msg_id, retry_count, error, next_retry_at): ...
    async def mark_dead(self, msg_id, retry_count, error): ...

worker = RetryWorker(repo=MyRepo(), client=bd)
await worker.start()

# 5. Admin router (optional) — adds operator endpoints.
class MyDirectory:
    async def list_agents(self): ...
    async def overview_snapshot(self): ...
    async def activate(self, agent_id): ...
    async def deactivate(self, agent_id): ...

app.include_router(
    make_router(client=bd, repo=MyRepo(), directory=MyDirectory()),
    prefix="/api/butt-dial",
)

---

Diagnostic: buttdial doctor

Seven staged checks with remediation for common failure modes:

$ buttdial doctor --to [phone]
[✓] Config loaded — url=https://...
[✓] Server reachable (142ms) — HTTP 200
[✓] SSE handshake (310ms)
[✓] Tool list — 7 tool(s), includes comms_send_message
[✗] send_message accepted (656ms) — agentId is required (or use a client token)
    → This tool needs agent context. Pass --agent-id <per-agent token>
      to the CLI, or send with agent_token=... from code.

Flags:

• --to +E.164 — recipient for stages 5-7
• --agent-id <token> — per-agent token for tools that require it (since 0.1.1)
• --ascii — use [OK] / [FAIL] instead of Unicode checkmarks (auto on Windows Git Bash)
• --expect-ack — wait for human reply on stage 7

Each failure includes a one-line remediation drawn from a curated map: 401/403/429, connection refused, timeouts, DNS, SSL, invalid recipient, missing token, agentId required, consent, rate limit, unprovisioned channel. Exit code is non-zero on any stage failure — wire it into CI as a live integration canary.

Server-returned errors (e.g. {"error": "..."} response bodies) are surfaced verbatim into SendResult.error — no more "no message_sid returned" silence (fixed in 0.1.1).

---

Testing with FakeButtDialServer

No real server required:

from buttdial import Agent
from buttdial.testing import fake_server   # pytest fixture

async def test_my_integration(fake_server):
    fake_server.on_send(sid="SM-42")
    c = Agent(base_url="https://fake.test", agent_id="agt-1", agent_token="t")
    result = await c.send_message(to="+1", body="hi")
    assert result.message_sid == "SM-42"
    assert fake_server.sent_messages[0].args["to"] == "+1"

Program error paths:

fake_server.on_send(error="connection refused", count=3)  # exhausts retries
fake_server.on_send(error="transient", count=2)           # recovers on 3rd
fake_server.on_send(sid="SM-recovered")

Simulate inbound events end-to-end:

await fake_server.simulate_inbound(inbound, sender="+1", text="hello", channel="whatsapp")
await fake_server.simulate_receipt(inbound, message_sid="SM-42", status="delivered")

---

Design principles

• Stateless — SDK owns no DB, no files, no global state beyond a configured Agent / AccountsClient.
• Protocols, not base classes — host implements FailedMessageRepository, AgentDirectory as Protocols. SDK never queries the DB.
• Errors are reported, not raised — SendResult.failed/error/stage_failed/remediation is the primary surface. The few places the SDK raises inherit from ButtDialError.
• Async-first — built on asyncio, httpx, mcp.
• Easy to test — FakeButtDialServer covers every MCP call; no port allocation, no external process.

---

Documentation

• docs/SPEC.md — full architecture and public API contract.
• docs/TODO.md — implementation roadmap (11 phases; 1-10 done).
• docs/DECISIONS.md — design decision log with rationale.
• docs/ERRORS.md — known pitfalls and their remediations.
• docs/REASONING.md — debugging breadcrumb trails.
• CHANGELOG.md — release notes.

---

License

MIT — free for commercial use, no obligations.