Python SDK

pip install bluebot

import os
from bluebot import BlueBot

client = BlueBot(api_key=os.environ["BLUEBOT_API_KEY"])

with BlueBot() as client:
    client.messages.send(to="+15555550123", text="Your table for four is ready.")

message = client.messages.send(
    to="+15555550123",
    text="Hi Maria, the showing at 412 Oak St is confirmed for 3 PM.",
    metadata={"crm_id": "48213"},
)

print(message.id, message.state)  # msg_1Hq2LxW9cVbN3mKpRtYs0a queued
if message.hold:
    print(f"Waiting: {message.hold.description}")

data = message.model_dump(mode="json", by_alias=True)

from datetime import datetime, timedelta, timezone

# Schedule one. It stays a draft you can change until it's due.
reminder = client.messages.send(
    to="+15555550123",
    text="Reminder: your cleaning is tomorrow at 10 AM. Reply C to confirm.",
    send_at=datetime(2026, 11, 20, 15, 0, tzinfo=timezone.utc),
    metadata={"appointment_id": "apt_9921"},
)

# Move it, or change the words, before it goes.
client.messages.update(reminder.id, send_at="2026-11-21T15:00:00Z")

# Or cancel it. A text the Mac already took can't be called back: that's a ConflictError.
client.messages.cancel(reminder.id)

# Text an existing group chat instead of one person.
client.messages.send(chat_id="chat_1Hq2LxW9cVbN3mKpRtYs0a", text="Running 10 minutes late.")

# Insist on iMessage (no SMS fallback), and give up if it hasn't gone in an hour.
client.messages.send(
    to="maria@example.com",
    text="Here is the link you asked for.",
    service="imessage",
    expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
)

# Read one.
sent = client.messages.get("msg_1Hq2LxW9cVbN3mKpRtYs0a")

# Settle a text the Mac couldn't confirm: you checked and it went, or send it again.
client.messages.resolve("msg_1Hq2LxW9cVbN3mKpRtYs0a", action="mark_sent")

from pathlib import Path

photo = client.files.upload(Path("house.jpg"))
client.messages.send(to="+15555550123", text="Here is the front of the house.", files=[photo.id])

import httpx

flyer = httpx.get("https://example.com/flyer.png")
upload = client.files.upload(flyer.content, filename="flyer.png")

# One page.
page = client.messages.list(direction="in", limit=50)
print(len(page.data), page.has_more, page.next_cursor)

# Every item, page after page.
for message in client.messages.list(state="failed", since="2026-11-01T00:00:00Z"):
    print(message.to, message.failure.reason if message.failure else None)

# By hand.
page = client.chats.list(is_group=True)
while True:
    for chat in page.data:
        print(chat.name)
    if not page.has_next_page():
        break
    page = page.get_next_page()

lookup = client.lookup("+15555550123")
print(lookup.service, lookup.basis)  # imessage history

device = client.device.get()
if device.status != "online":
    print(f"Offline since {device.offline_since}")
for name, check in device.checks:
    if check.state == "needs_you":
        print(f"{name}: {check.detail}")
print(f"{device.today.new_people} of {device.today.new_people_limit} new people today")

status = client.status()
me = client.me()
print(status.sending, me.account.email, me.key.scopes if me.key else None)

settings = client.settings.update(
    time_zone="America/Chicago",
    pace_seconds=30,
    quiet_hours={"enabled": True, "start": "20:00", "end": "08:00"},
)
print(settings.warnings)

# Pause all sending for two hours, then resume.
client.settings.update(paused_until=datetime.now(timezone.utc) + timedelta(hours=2))
client.settings.update(paused_until=None)

client.stops.add(person="+15555550144")
for stop in client.stops.list():
    print(stop.handle, stop.source)
client.stops.remove("+15555550144")  # the SDK encodes the + for the path

for event in client.events.list(type=["message.received", "message.failed"], since="2026-11-12T00:00:00Z"):
    print(event.type, event.timestamp)

event = client.events.get("evt_1Hq2LxW9cVbN3mKpRtYs0a")

webhook = client.webhooks.create(
    url="https://example.com/bluebot/webhooks",
    events=["message.received", "message.failed"],
    description="Replies into the CRM",
)
print(webhook.secret)  # shown now and on rotation, never again: store it

client.webhooks.test(webhook.id, type="message.received")
client.webhooks.update(webhook.id, paused=True)
secret = client.webhooks.rotate_secret(webhook.id).secret

for delivery in client.webhooks.deliveries.list(webhook.id, state="gave_up"):
    client.webhooks.deliveries.replay(webhook.id, delivery.id)

client.webhooks.delete(webhook.id)

import os
from flask import Flask, abort, request
from bluebot import BlueBot, WebhookVerificationError

app = Flask(__name__)
client = BlueBot()


@app.post("/bluebot/webhooks")
def bluebot_webhook():
    try:
        event = client.webhooks.verify(
            request.get_data(),  # the raw body, byte for byte
            request.headers,
            secret=os.environ["BLUEBOT_WEBHOOK_SECRET"],
        )
    except WebhookVerificationError:
        abort(400)

    if event.type == "message.received":
        message = event.data  # the message
        print(f"{message.from_}: {message.text}")
    return "", 204

import os
from fastapi import FastAPI, HTTPException, Request, Response
from bluebot import BlueBot, WebhookVerificationError

app = FastAPI()
client = BlueBot()


@app.post("/bluebot/webhooks")
async def bluebot_webhook(request: Request):
    try:
        event = client.webhooks.verify(await request.body(), request.headers, secret=os.environ["BLUEBOT_WEBHOOK_SECRET"])
    except WebhookVerificationError:
        raise HTTPException(status_code=400)

    if event.type == "contact.stopped":
        print(f"{event.data.handle} stopped")
    return Response(status_code=204)

import asyncio
from bluebot import AsyncBlueBot


async def main():
    async with AsyncBlueBot() as client:
        message = await client.messages.send(to="+15555550123", text="Your order shipped today.")
        print(message.state)

        async for received in client.messages.list(direction="in", limit=100):
            print(received.from_, received.text)

        page = await client.messages.list(state="failed")
        print(len(page.data))


asyncio.run(main())

client.messages.send(
    to=lead.phone,
    text=f"Hi {lead.first_name}, thanks for your interest.",
    idempotency_key=f"welcome-{lead.id}",
)

raw = client.messages.with_raw_response.send(to="+15555550123", text="Hello again.", idempotency_key="hello-again-1")
print(raw.headers.get("Request-Id"), raw.headers.get("Idempotent-Replayed"))
message = raw.parse()

import bluebot

try:
    client.messages.send(to="555", text="Hi")
except bluebot.BadRequestError as error:
    for detail in error.details or []:
        print(f"{detail.field}: {detail.message}")
except bluebot.RateLimitError as error:
    print(f"Slow down; try again in {error.retry_after} seconds")
except bluebot.APIError as error:
    print(f"{error.code}: {error.message} {error.hint} ({error.request_id})")
except bluebot.APIConnectionError:
    print("BlueBot could not be reached")

from bluebot.types import Message, Hold, FileObject, Chat, Contact, Lookup, Device, Settings, Webhook, WebhookDelivery
from bluebot.types import WebhookEvent, MessageReceivedEvent, ContactStoppedEvent

import httpx
from bluebot import BlueBot


def handler(request: httpx.Request) -> httpx.Response:
    return httpx.Response(201, json={"id": "msg_1Hq2LxW9cVbN3mKpRtYs0a", "object": "message", "state": "queued"})


client = BlueBot(api_key="bb_test_unused", max_retries=0, http_client=httpx.Client(transport=httpx.MockTransport(handler)))