butt-dial-sdk 0.1.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 in a single installable package.

Status: Alpha (0.1.0). Already in production use via iv-bknd. Not yet published to PyPI — install from source.

---

Install

pip install 'butt-dial-sdk @ git+https://github.com/95percent-ai/butt-dial-sdk.git'
# Or from source:
git clone https://github.com/95percent-ai/butt-dial-sdk.git
pip install -e ./butt-dial-sdk
# With FastAPI router support:
pip install -e './butt-dial-sdk[router]'

Quickstart — send a message in 3 lines

Set two env vars:

export BUTT_DIAL_URL=https://your-server.example/sse
export BUTT_DIAL_TOKEN=your-org-token

Send:

from buttdial import Client

bd = Client.from_env()
result = await bd.send_message(to="[phone]", body="Hello from Python")
print(result.message_sid)  # "SM..." on success

That's it. 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
Client	Outbound send_message(), list_tools(), ping(), fetch_usage_summary(). MCP/SSE transport with 3-attempt exponential-backoff retry and token resolution (agent > org > legacy).
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.

---

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 (
    Client, InboundHandler, RetryWorker,
    FailedMessageRepository,    # protocol — host implements
    AgentDirectory,             # protocol — host implements
    make_router,
)

# 1. Build a client (once, at startup).
bd = Client.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 (820ms) — sid=SM_xyz
[✗] Delivery receipt — no receipt for SM_xyz within 30s
    → Inbound webhook did not deliver a receipt. Verify POST /{channel}
      is reachable from Butt-Dial and that the server is configured to
      POST delivery receipts to this app.

Each failure includes a one-line remediation drawn from a curated map of known failure modes (401/403, connection refused, timeouts, DNS, SSL, invalid recipient, missing token). Exit code is non-zero when any stage fails — wire it into CI as a live integration canary.

---

Testing with FakeButtDialServer

No real server required:

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

async def test_my_integration(fake_server):
    fake_server.on_send(sid="SM-42")
    c = Client(token="t", url="https://fake.test/sse")
    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 Client.
• 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.