butt-dial-sdk 0.4.0

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.3.0. In production via iv-bknd.

---

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+. 170/170 tests green.

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.

---

Connect butt-dial

Onboard a new tenant onto Butt-Dial and get their first owner bearer token. Once this cycle is done, the rest of provisioning (buying phones / emails / SMS numbers, activating IM devices) happens through Agent and the admin console:

from buttdial import AccountsClient

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

# 1. Register — the server emails a verification link.
await accounts.register("Acme Inc", "[e-mail]")
# … user clicks the link; your app's deep-link handler extracts the token …

# 2. Claim the team token.
info = await accounts.verify(verification_token)
vault.put(tenant_id, info["team_token"])
authed = accounts.with_token(info["team_token"])

# 3. Register an admin phone (prerequisite for rotation).
challenge = await authed.set_phone("[phone]")
await authed.confirm_phone(challenge["challenge_id"], otp_from_user)

# 4. Later — rotate. The old token is revoked atomically.
new = await authed.rotate_token(otp_prompt=lambda hint:
    input(f"Enter code sent to {hint}: "))
vault.put(tenant_id, new["token"])

---

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.
• docs/CHANGELOG.md — release notes.

---

License

MIT — free for commercial use, no obligations.