Handle iMessage Events in Python with FastAPI

Figment's Apple Connect exposes iMessage through a REST API — read, search, and send. It does not push incoming messages to your server. Building a reactive Python pipeline means writing a poller that calls Figment's messages endpoint on a schedule, detects new arrivals by GUID, and routes them to your FastAPI handlers. This guide delivers the full design: a polling loop, an HMAC-signed internal event receiver, and durable deduplication that survives process restarts.

Install dependencies

uv add "fastapi[standard]" httpx
# Add redis-py only if you use the production deduplication variant below
uv add redis

How the two services connect

The design has two separate processes: a poller that holds the Figment access token and calls the messages endpoint on a schedule, and a FastAPI handler that receives signed messages from the poller and dispatches them to your application logic.

The signing contract — the shared secret, header name, and POST envelope — is your application's design. The poller and handler share a secret you generate and store; that secret never leaves your infrastructure and is not a Figment-issued signing scheme. This keeps the handler testable in isolation: you can POST mock payloads to it locally without involving the Figment API at all.

Figment's sandbox environment serves synthetic messages, so the entire loop — polling, signing, verifying, deduplicating — can be exercised without a live Apple Account connection. Only the connection ID changes between sandbox and live.

Discover which chats to poll

Before polling messages, your application needs the chatGuid values it will watch. GET /v1/connections/{connection_id}/chats returns a paginated list of conversations; use the guid field from each chat item as the chatGuid parameter when fetching messages. Store those values in your configuration or database; the polling loop iterates over each chatGuid in turn. The messages:read scope covers both chat listing and message history.

The poller

# poller.py — your application's polling service
import asyncio
import hashlib
import hmac
import json
import os

import httpx

BASE_URL        = "https://figment.so/api/apple-connect"
CONNECTION_ID   = os.environ["FIGMENT_CONNECTION_ID"]    # e.g. con_...
# Access tokens expire after ~900 s; refresh via the OAuth flow before each cycle.
ACCESS_TOKEN    = os.environ["FIGMENT_ACCESS_TOKEN"]
CHAT_GUID       = os.environ["FIGMENT_CHAT_GUID"]        # from GET .../chats
INTERNAL_URL    = os.environ["INTERNAL_HANDLER_URL"]     # e.g. http://localhost:8000/internal/imessage-event
INTERNAL_SECRET = os.environ["INTERNAL_WEBHOOK_SECRET"]  # your application secret


async def poll_once(client: httpx.AsyncClient, state: dict) -> None:
    """Fetch one page of messages; emit new ones to the internal handler."""
    params: dict = {"chatGuid": CHAT_GUID, "limit": 20}
    if state.get("next_offset") is not None:
        params["offset"] = state["next_offset"]

    # httpx percent-encodes params, correctly handling semicolons and plus signs
    # in chatGuid values such as "iMessage;+;+15551234567".
    r = await client.get(
        f"{BASE_URL}/v1/connections/{CONNECTION_ID}/messages",
        params=params,
        headers={"Authorization": f"Bearer {ACCESS_TOKEN}"},
    )
    r.raise_for_status()   # non-2xx (e.g. non-ready connection) raises an exception

    envelope = r.json()
    # envelope shape: {data, mode, offset, nextOffset, hasMore, historyComplete, scope}
    for msg in envelope.get("data", []):
        # msg shape: {guid, text, isFromMe, chatGuid}
        await emit_event(client, msg)

    if envelope.get("hasMore") and envelope.get("nextOffset") is not None:
        state["next_offset"] = envelope["nextOffset"]
    else:
        # hasMore is False or nextOffset is absent: we are at the current edge.
        # Reset so the next cycle starts at offset 0 to pick up newly arrived messages.
        state.pop("next_offset", None)


async def emit_event(client: httpx.AsyncClient, msg: dict) -> None:
    """Sign and POST one message object to the internal handler."""
    body = json.dumps(msg).encode()
    sig = hmac.digest(INTERNAL_SECRET.encode(), body, hashlib.sha256).hex()
    await client.post(
        INTERNAL_URL,
        content=body,
        headers={"Content-Type": "application/json", "X-Internal-Signature": sig},
    )


async def run_poller(interval: int = 15) -> None:
    state: dict = {}
    async with httpx.AsyncClient() as client:
        while True:
            try:
                await poll_once(client, state)
            except Exception as exc:
                print(f"poll error: {exc}")
            await asyncio.sleep(interval)


if __name__ == "__main__":
    asyncio.run(run_poller())

Track next_offset in a durable store (database or Redis) for multi-instance deployments; the in-dict state above resets on restart. hasMore: false means you are at the current edge of available history. historyComplete: false means the initial sync window is not yet complete — earlier messages may not yet be present, so poll again rather than treating hasMore: false as proof of full coverage.

Access tokens expire after approximately 900 seconds. Before each call to the messages endpoint, check whether your stored token is about to expire and refresh it using the OAuth refresh flow. After refreshing, persist the newly issued refresh token before the old one is invalidated; refresh tokens rotate. Access tokens are opaque strings — do not decode them as JWTs.

Your FastAPI event handler

# handler.py — your application's internal event receiver
import hashlib
import hmac
import json
import os
from typing import Annotated

from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request

app = FastAPI()

INTERNAL_SECRET = os.environ["INTERNAL_WEBHOOK_SECRET"]
_seen_guids: set[str] = set()   # development only — replace in production (see below)


@app.post("/internal/imessage-event", status_code=200)
async def imessage_event(
    request: Request,
    background_tasks: BackgroundTasks,
    x_internal_signature: Annotated[str | None, Header()] = None,
):
    body = await request.body()
    _verify_signature(body, x_internal_signature)

    msg = json.loads(body)
    guid = msg.get("guid", "")
    if _is_duplicate(guid):
        return {"status": "duplicate"}

    if not msg.get("isFromMe"):
        background_tasks.add_task(_dispatch_incoming, msg)

    return {"status": "accepted"}


def _dispatch_incoming(msg: dict) -> None:
    # Route to your application logic here.
    # Avoid recording message text in audit trails.
    print(f"[{msg.get('chatGuid')}] incoming guid={msg.get('guid')}")


def _verify_signature(payload: bytes, header_value: str | None) -> None:
    """Your application's signing contract — not a Figment-issued signature."""
    # INTERNAL_SECRET is loaded at startup via os.environ[]; the server refuses
    # to start if the variable is absent, so there is no silent bypass here.
    if header_value is None:
        raise HTTPException(status_code=401, detail="Missing signature")
    expected = hmac.digest(
        INTERNAL_SECRET.encode(), payload, hashlib.sha256
    ).hex()
    if not hmac.compare_digest(expected, header_value):
        raise HTTPException(status_code=401, detail="Signature mismatch")


def _is_duplicate(guid: str) -> bool:
    """Return True if this guid was already processed; mark it if not."""
    if not guid or guid in _seen_guids:
        return True
    _seen_guids.add(guid)
    return False

The isFromMe check filters out messages the connected account sent outbound. Sent messages appear in the same polling results; without the filter, _dispatch_incoming would run for every outbound message too, risking duplicate CRM writes or unintended reply loops. Remove the check if your workflow needs to track both directions.

BackgroundTasks returns 200 OK before _dispatch_incoming completes, so the poller does not wait on a slow database write before treating the event as acknowledged. FastAPI's BackgroundTasks has no completion guarantee if the process exits before the task runs (FastAPI background tasks docs). For writes that must not be lost — CRM updates, outbound replies — push the message GUID into a durable queue (Celery, ARQ, or a database job table) inside _dispatch_incoming and do the work there.

hmac.compare_digest runs in constant time regardless of where the strings first differ, eliminating a timing side-channel (Python hmac docs). Never compare HMAC signatures with ==.

Durable deduplication for production

The in-process set in handler.py is not shared across workers and does not survive a restart. Replace _is_duplicate with a Redis SETNX call:

# Requires redis-py: uv add redis
import redis
_redis = redis.Redis.from_url(os.environ["REDIS_URL"])

def _is_duplicate(guid: str) -> bool:
    # set(..., nx=True) returns True when the key is newly set (not a duplicate).
    created = _redis.set(f"imsg:seen:{guid}", 1, nx=True, ex=86400)
    return not created

A database upsert with a unique constraint on guid is an equally valid alternative when you are already writing message records. The 24-hour TTL covers typical retry windows; expand it if late retries are possible in your environment.

For outbound sends from your application, generate a unique key — such as a UUID4 — per intended send action and persist it in a durable store alongside the recipient, text, and connection ID before the first API request. Reuse that exact stored key when retrying. Do not derive the key from a hash of the message content alone: two separate intended sends of the same text to the same recipient would share a content hash and be wrongly collapsed to one delivery.

Run the two services

# Terminal 1 — FastAPI handler
uv run uvicorn handler:app --reload

# Terminal 2 — poller
uv run python poller.py

For production, run each as a managed process (systemd unit, container, or supervisor) so both restart automatically. The durable deduplication store means a restarted poller will not reprocess messages it has already delivered.

Connect and configure

The messages:read scope is required for the poller; messages:send is a separate scope required only if your application sends replies. Register your developer application and begin the PKCE authorization flow at Apple Connect for Developers.

The iMessage connection must reach ready status before the messages endpoint returns data. Use the connection states table below to decide how the poller should react, and monitor status at /apple-connect/account during setup.

When Messages in iCloud is enabled on the connected account and an eligible iPhone is on and online, the available history includes SMS, MMS, and RCS conversations — not only iMessage traffic (Apple support).

A live connection costs $29 per month; sandbox access is free. Access starts after setup.

Connect Apple Account to begin the authorization flow.