iMessage API for Python

Building on the iMessage API? Build on iMessage with Flow: one agent for iMessage, WhatsApp and Telegram. New to it? Read the iMessage API guide.

Install and get a key

pip install flow-messaging
npx @flow-engineer/messaging init

You need Python 3.10 or later. init creates a test key with no account and writes FLOW_MESSAGING_KEY to .env. Without Node, call the endpoint yourself:

import requests
print(requests.post("https://api.flow.engineer/v1/sandbox/keys").json())

Save the key it returns as FLOW_MESSAGING_KEY. It is shown once. The test key runs on the Telegram sandbox. The same handler answers on iMessage, on Flow’s shared or dedicated US numbers.

Stream and reply

This is the Python example from Flow’s quickstart:

from flow_messaging import Flow, content_text

flow = Flow()
for event in flow.events.stream(types=["message.received"]):
    said = content_text(event.data.message.content)
    flow.messages.send(
        event.conversation.id,
        f"You said: {said}",
        idempotency_key=event.id,
    )

Replace the echo with a call to your model. A send is queued first, so watch message.sent or message.failed to learn the outcome. Flow has Python examples for the OpenAI Agents SDK and LangChain.

A FastAPI webhook

In production, register an HTTPS endpoint and verify every delivery. This is Flow’s FastAPI example:

from fastapi import FastAPI, Request, Response
from flow_messaging import WebhookSignatureError, construct_event, content_text, webhook_reply

app = FastAPI()

@app.post("/flow")
async def flow_webhook(request: Request):
    body = await request.body()  # the raw body: a parsed and re-serialized body will not match
    try:
        # The secret defaults to FLOW_MESSAGING_WEBHOOK_SECRET.
        event = construct_event(body, request.headers.get("Flow-Signature"))
    except WebhookSignatureError:
        return Response(status_code=400)
    if event.type == "message.received":
        return webhook_reply(f"You said: {content_text(event.data.message.content)}")
    return {}

Three rules: use the raw body, answer within 10 seconds or return {} and send later, and deduplicate on the event id, because deliveries are at least once. iMessage webhooks: signing, retries and ordering covers the details.

The raw WebSocket stream

In any language, you can read the live stream with a WebSocket client that can set a header. This is the example from Flow’s local development guide:

# pip install websockets
import asyncio, json, os, websockets

async def main():
    url = "wss://api.flow.engineer/v1/stream?type=message.received"
    headers = {"Authorization": f"Bearer {os.environ['FLOW_MESSAGING_KEY']}"}
    async with websockets.connect(url, additional_headers=headers) as ws:
        async for raw in ws:
            frame = json.loads(raw)
            print(frame)  # an "event" frame; reply with POST /v1/conversations/{id}/messages

asyncio.run(main())

Keep the id of the last event you handled. When you reconnect, pass it as after and the stream replays what you missed, then goes live.

Other languages: Node.js and TypeScript and Go.

Frequently asked questions

Keep reading

All posts: the Flow blog. The Flow docs have the reference.