# Read and Search iMessages with Python

Canonical: https://figment.so/blog/read-imessages-python

Your Python code can list conversations, walk message history, and run full-text searches over iMessage — all through the Figment Apple Connect REST API. Standard HTTP, the `requests` library, a bearer token: the developer app receives a scoped access token, never an Apple password, and your code runs on any platform.

## How iMessage history reaches your code

The Messages app on the connected Mac is the read source. When Messages in iCloud is enabled, Apple syncs iMessage, SMS, MMS, and RCS history across devices configured for Messages in iCloud under the same Apple Account ([Apple support HT102545](https://support.apple.com/en-us/102545)). Without Messages in iCloud, Text Message Forwarding can be configured separately on the iPhone. The iPhone must be on and connected for new carrier traffic to arrive; previously synced carrier history can remain on the Mac while the phone is offline. The bridge reads the local database on the Mac; your server never touches the SQLite file directly.

A Mac that has just reconnected may still be downloading older history. The API exposes a per-service status — `connecting`, `needs_approval`, `syncing`, `ready`, `offline`, `reconnect_required`, or `revoked` — so you can guard against premature reads. Only a `ready` service accepts message operations; a non-ready service **rejects** them. Even a `ready` service may have incomplete history if the account was recently reconnected; check `historyComplete` in the list envelope rather than assuming full coverage.

## Authorize your app

[Register your app](https://figment.so/apple-connect/developers) at the developer dashboard, then walk the user through Figment's PKCE consent screen. This grants your app scoped access to an existing Apple connection — it is not the Apple sign-in step itself, which the account owner completed during connection setup.

Generate a **fresh** code verifier and challenge for every authorization request. Send the user to:

```
GET https://figment.so/apple-connect/authorize
  ?response_type=code
  &client_id=YOUR_CLIENT_ID
  &redirect_uri=https://yourapp.example/callback
  &scope=messages:read
  &state=RANDOM_OPAQUE_VALUE
  &code_challenge=CODE_CHALLENGE
  &code_challenge_method=S256
```

Validate the returned `state` before exchanging the code, then POST to the token endpoint:

```
POST https://figment.so/api/apple-connect/oauth/token
Content-Type: application/x-www-form-urlencoded

grant_type=authorization_code&client_id=YOUR_CLIENT_ID&redirect_uri=https%3A%2F%2Fyourapp.example%2Fcallback&code=RETURNED_CODE&code_verifier=YOUR_CODE_VERIFIER
```

The response includes `access_token`, `refresh_token`, `expires_in` (900 seconds for the access token), `scope`, and `connection_id`. Store the `connection_id` — a `con_…` value — alongside both tokens. To refresh, POST `grant_type=refresh_token`, `client_id`, and `refresh_token` as a serialized `application/x-www-form-urlencoded` body to the same token endpoint — one refresh at a time, per grant; always persist the new `refresh_token` from each response, as tokens rotate on use.

Export the credentials and install the HTTP client:

```bash
export FIGMENT_ACCESS_TOKEN=ACCESS_TOKEN
export FIGMENT_CONNECTION_ID=con_01abc...
pip install requests
```

Manage existing grants at [figment.so/apple-connect/account](https://figment.so/apple-connect/account).

## Check the connection and list chats

Call `GET /v1/connections/{connection_id}` to confirm the service is ready before reading. This endpoint returns `{connection:{id,mode,services,...}}` — a different shape from the list endpoints — and can return a non-2xx error if the connection is unavailable or revoked. Handle that before proceeding:

```python
import os, sys, requests

BASE_URL      = "https://figment.so/api/apple-connect"
ACCESS_TOKEN  = os.environ["FIGMENT_ACCESS_TOKEN"]
CONNECTION_ID = os.environ["FIGMENT_CONNECTION_ID"]

HEADERS = {"Authorization": f"Bearer {ACCESS_TOKEN}"}


def get_connection():
    r = requests.get(
        f"{BASE_URL}/v1/connections/{CONNECTION_ID}",
        headers=HEADERS, timeout=10,
    )
    r.raise_for_status()
    return r.json()["connection"]


def list_chats(limit=20, offset=0):
    r = requests.get(
        f"{BASE_URL}/v1/connections/{CONNECTION_ID}/chats",
        headers=HEADERS,
        params={"limit": limit, "offset": offset},
        timeout=10,
    )
    r.raise_for_status()
    return r.json()


conn = get_connection()

# Each service entry is an object: {state, reason, observedAt}.
# Only granted services appear in the services map.
messages_svc = conn.get("services", {}).get("messages", {})
state = messages_svc.get("state")
if state != "ready":
    sys.exit(
        f"Messages service not ready (state={state!r}). "
        "Check https://figment.so/apple-connect/account"
    )

page = list_chats(limit=20)
chats = page["data"]
if not chats:
    sys.exit("No chats returned.")

print(f"Mode: {page['mode']} — {len(chats)} chat(s) on this page")
```

The `mode` field in every list response is `"live"` or `"sandbox"`. Sandbox responses contain only synthetic data; nothing reaches real recipients.

## Paginate through messages

Use each chat item's `guid` as the `chatGuid` parameter to the messages endpoint. Drive the loop with `hasMore` and `nextOffset` — not a length comparison — and stop before the documented offset limit:

```python
def list_messages(chat_guid, limit=20, offset=0):
    r = requests.get(
        f"{BASE_URL}/v1/connections/{CONNECTION_ID}/messages",
        headers=HEADERS,
        params={"chatGuid": chat_guid, "limit": limit, "offset": offset},
        timeout=15,
    )
    r.raise_for_status()
    return r.json()


MAX_OFFSET = 10000  # inclusive public offset limit; nextOffset > 10000 must stop
first_guid = chats[0]["guid"]   # each chat item's guid

all_messages, offset, stop_reason = [], 0, "hasMore=False"
while True:
    envelope = list_messages(first_guid, limit=20, offset=offset)
    all_messages.extend(envelope["data"])
    if not envelope.get("hasMore"):
        stop_reason = "hasMore=False"
        break
    next_offset = envelope.get("nextOffset")
    if next_offset is None:
        stop_reason = "nextOffset=None"
        break
    if next_offset <= offset:
        stop_reason = f"non-advancing nextOffset ({next_offset})"
        break
    if next_offset > MAX_OFFSET:
        stop_reason = f"nextOffset {next_offset} exceeds public limit"
        break
    offset = next_offset

print(f"Loaded {len(all_messages)} message(s). Stopped: {stop_reason}.")
# historyComplete in the last envelope reflects what the Mac's local store covered.
# It does not prove all iCloud history is present locally.
```

The `requests` library encodes the `params` dict automatically — never build the query string by hand with values containing colons or plus characters. Message items include `guid`, `text`, `isFromMe`, and `chatGuid`; `displayName` on a chat item may be absent. Empty `data` arrays are valid responses.

## Search messages

History reads and full-text searches share the single `messages:read` scope — there is no separate search permission. Supply `query`; optionally scope results to one conversation with `chatGuid`. With a selected-conversation grant, `chatGuid` is required:

```python
def search_messages(query, chat_guid=None, limit=20, offset=0):
    params = {"query": query, "limit": limit, "offset": offset}
    if chat_guid:
        params["chatGuid"] = chat_guid
    r = requests.get(
        f"{BASE_URL}/v1/connections/{CONNECTION_ID}/messages/search",
        headers=HEADERS,
        params=params,
        timeout=20,
    )
    r.raise_for_status()
    return r.json()


results = search_messages("invoice", chat_guid=first_guid, limit=20)
print(f"Found {len(results['data'])} result(s). hasMore: {results.get('hasMore')}")
```

Search scans the messages the Mac has downloaded locally — it is not a guaranteed complete full-text index of all-time history. Pagination follows the same `hasMore` / `nextOffset` pattern and the same `offset` ≤ 10000 bound.

## Sync state, SMS, and RCS

**Non-ready services reject reads.** The connection endpoint can reject a non-ready or revoked connection with a non-2xx error rather than returning a state object; handle that with `raise_for_status` as the code above does. If the service is `syncing`, `offline`, or `reconnect_required`, message and search calls will be rejected. For connection setup status, go to [figment.so/apple-connect/account](https://figment.so/apple-connect/account). Once the service is ready, check `historyComplete` in the list envelope — it reflects what the Mac's local store covered, not whether all iCloud history is present locally.

**SMS, MMS, and RCS.** When Messages in iCloud is enabled on the user's iPhone, carrier message types sync to the Mac automatically alongside iMessage ([Apple HT102545](https://support.apple.com/en-us/102545)). Without Messages in iCloud, Text Message Forwarding can be configured separately. The iPhone must be on and connected for new carrier traffic to arrive; previously synced carrier history can remain on the Mac while the phone is offline.

**Reconnect interruptions.** A `reconnect_required` state means account status needs attention; there is no single asserted cause. Direct users to [figment.so/apple-connect/account](https://figment.so/apple-connect/account) rather than returning stale data as current.

---

**Sending.** `messages:read` covers both history reads and full-text search. Sending requires the separate `messages:send` scope and an `Idempotency-Key` header on every POST. A chat visible in read results is not automatically available for sends; the send grant must include that specific existing conversation.

**Pricing.** The developer pays **$29/month** per active live connection. Sandbox is free, and access starts after setup.

---

For app registration and the full OAuth reference, start at the [developer dashboard](https://figment.so/apple-connect/developers).

**[Connect Apple Account](https://figment.so/apple-connect/developers)** to link your iMessage history to your Python app.