Skip to content
imessageapi

iMessage API for Python

Python has no special standing with any iMessage provider and does not need one — every vendor is an HTTP call. Here is the client, the retry policy, the FastAPI webhook and the pandas-shaped bulk send, written the way you would actually ship them.

8 min readUpdated August 24, 2026Language

There is no Python-specific difficulty in this category. Every hosted provider is REST, so httpx is the integration. What follows is the surrounding code that makes it production-shaped rather than a snippet.

A client worth keeping

One class, one place that knows the vendor. When you switch providers — and this is a young category, so plan for it — only this file changes.

messaging.py
import os
from dataclasses import dataclass
 
import httpx
 
 
class SendError(Exception):
    def __init__(self, message: str, status: int, retryable: bool):
        super().__init__(message)
        self.status = status
        self.retryable = retryable
 
 
@dataclass(frozen=True)
class SendResult:
    id: str
    channel: str
 
 
class Messaging:
    def __init__(self, base_url: str | None = None, api_key: str | None = None):
        self._client = httpx.Client(
            base_url=base_url or os.environ["IMESSAGE_API_URL"],
            headers={"Authorization": f"Bearer {api_key or os.environ['IMESSAGE_API_KEY']}"},
            # A hung provider must not hold your worker forever.
            timeout=httpx.Timeout(10.0, connect=5.0),
        )
 
    def send(self, to: str, text: str) -> SendResult:
        response = self._client.post("/v1/messages", json={"to": to, "text": text})
 
        if response.is_error:
            # 429 and 5xx are worth another go; 4xx means the call itself is wrong.
            retryable = response.status_code == 429 or response.status_code >= 500
            raise SendError(response.text, response.status_code, retryable)
 
        data = response.json()
        return SendResult(id=data["id"], channel=data.get("channel", "imessage"))
 
    def close(self) -> None:
        self._client.close()

Retries and rate limits

Providers throttle on purpose — hammering a line is what gets the underlying Apple account flagged. Treat a 429 as useful advice rather than an obstacle, and honour Retry-After when the vendor sends it.

retry.py
import random
import time
 
from messaging import Messaging, SendError, SendResult
 
 
def send_with_retry(client: Messaging, to: str, text: str, attempts: int = 4) -> SendResult:
    for attempt in range(attempts):
        try:
            return client.send(to, text)
        except SendError as err:
            if attempt == attempts - 1 or not err.retryable:
                raise
            # Jitter stops a batch of failures from retrying in lockstep and
            # rebuilding the exact spike that triggered the throttle.
            base = (2 ** attempt) * 0.5
            time.sleep(base + random.random() * base)
 
    raise AssertionError("unreachable")

Receiving replies with FastAPI

Inbound webhooks are delivered at least once. Duplicates will happen, so deduplicate on the provider's message id and acknowledge before you do any real work.

webhook.py
import hashlib
import hmac
import json
import os
 
from fastapi import BackgroundTasks, FastAPI, Request, Response
 
app = FastAPI()
SECRET = os.environ["IMESSAGE_WEBHOOK_SECRET"].encode()
 
 
@app.post("/webhooks/imessage")
async def inbound(request: Request, background: BackgroundTasks) -> Response:
    # Verify against the exact bytes received, before any JSON round-trip.
    raw = await request.body()
    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
 
    if not hmac.compare_digest(expected, request.headers.get("x-signature", "")):
        return Response(status_code=401)
 
    event = json.loads(raw)
 
    # Return immediately; do the work after. A slow handler becomes a provider
    # retry, which becomes a duplicate message to your customer.
    background.add_task(handle, event)
    return Response(status_code=200)
 
 
async def handle(event: dict) -> None:
    message_id = event["message_id"]
    if await already_handled(message_id):
        return
    await mark_handled(message_id)
    await respond(event["from"], event["text"])

Bulk sending from a DataFrame

The most common Python use case in this category is 'I have a spreadsheet of customers and I want to text them'. Two things make the difference between that working and getting your line flagged: pace it, and personalise it.

campaign.py
import time
 
import pandas as pd
 
from messaging import Messaging
from retry import send_with_retry
 
 
def run_campaign(df: pd.DataFrame, template: str, per_minute: int = 30) -> pd.DataFrame:
    """Send one personalised message per row, paced to look human."""
    client = Messaging()
    gap = 60 / per_minute
    results = []
 
    try:
        for i, row in enumerate(df.itertuples()):
            text = template.format(first_name=row.first_name)
            try:
                result = send_with_retry(client, row.phone, text)
                results.append({"phone": row.phone, "id": result.id, "error": None})
            except Exception as err:  # one bad number must not end the run
                results.append({"phone": row.phone, "id": None, "error": str(err)})
 
            if i < len(df) - 1:
                time.sleep(gap)
    finally:
        client.close()
 
    return pd.DataFrame(results)

Check consent before the loop, not after

A DataFrame of phone numbers is not a list of people who agreed to hear from you. Filter on your consent column before you send, and honour opt-outs on every run — see consent before you text.

Reading iMessage directly on macOS

If your Python runs on a Mac, you can read the local Messages database without any vendor at all. Useful for analysis and for self-hosted inbound; read-only, and open a copy rather than the live file.

Reading chat.db read-only
import sqlite3
from pathlib import Path
 
DB = Path.home() / "Library" / "Messages" / "chat.db"
 
 
def recent_messages(limit: int = 50) -> list[dict]:
    # Read-only URI: never write to the live database Messages.app is using.
    # Requires Full Disk Access for your terminal or Python binary.
    conn = sqlite3.connect(f"file:{DB}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
 
    try:
        rows = conn.execute(
            """
            SELECT
                handle.id AS contact,
                message.text,
                message.is_from_me,
                -- Apple epoch is 2001-01-01, stored in nanoseconds.
                datetime(message.date / 1000000000 + 978307200, 'unixepoch', 'localtime') AS sent_at
            FROM message
            JOIN handle ON message.handle_id = handle.ROWID
            WHERE message.text IS NOT NULL
            ORDER BY message.date DESC
            LIMIT ?
            """,
            (limit,),
        ).fetchall()
    finally:
        conn.close()
 
    return [dict(row) for row in rows]

The Node equivalents are in iMessage API for Node.js. For the provider decision itself, start at who sells an iMessage API.

Common questions

What is the best iMessage API for Python?
All hosted providers work equally well from Python since they expose REST. Blooio publishes an official Python SDK alongside Node, Go and Java. Otherwise `httpx` or `requests` against the vendor's REST endpoint is the whole integration.
Is there a Python library for sending iMessages?
There are vendor SDKs on PyPI and open-source packages that read the local chat database on macOS. There is no official Apple library, because Apple publishes no API.

Related

All 8 pages