Give a LangGraph Agent Scoped iMessage Access

Three Python objects model the full permission boundary for an iMessage-aware LangGraph agent: a read_messages tool that queries the messages:read scope, a draft_reply tool that stages text in graph state without any network call, and a send_approved_message tool that requires both a separate messages:send grant and an explicit human-approval interrupt in the same StateGraph. The read and draft stages run freely inside the agentic loop; the send stage never fires until a reviewer resumes the paused graph. Figment's bridge sits between your agent and a Mac running iMessage — your code supplies a bearer access token and never handles the Apple password.

Integration design — The code examples in this article illustrate the pattern and have not been executed against a live Figment deployment. Use them as reference architecture alongside the iMessage API reference once your connection is active.

How Figment scopes iMessage access

Figment issues two distinct iMessage scopes. messages:read covers reading thread history and searching messages. messages:send is a separate scope that also requires the user to have explicitly selected the target conversation during Figment's consent flow — a conversation visible in read results is not automatically authorized for sends. The two scopes can be requested within one app grant and revoked independently.

This grant is Figment's own authorization layer, not Apple's documented third-party iCloud authorization (which covers only Mail, Calendar, and Contacts). The iMessage bridge runs on the signed-in Mac.

Before your agent reads anything, the connection must reach ready for the messages service. The connection endpoint GET /v1/connections/{connection_id} enforces authorization; a non-ready or revoked connection returns a non-2xx response rather than a diagnostic status object. Direct users to /apple-connect/account for connection setup. A ready state does not guarantee complete history or successful delivery.

Verify the connection before reading

A startup check surfaces a misconfigured connection before the agent enters its loop:

# startup check — integration design, not executed
import os, requests

BASE  = os.environ["FIGMENT_BASE_URL"]        # https://figment.so/api/apple-connect
TOKEN = os.environ["FIGMENT_ACCESS_TOKEN"]
CONN  = os.environ["FIGMENT_CONNECTION_ID"]   # con_... from your OAuth token response
HDR   = {"Authorization": f"Bearer {TOKEN}"}

resp = requests.get(f"{BASE}/v1/connections/{CONN}", headers=HDR)
if not resp.ok:
    raise RuntimeError(
        f"Connection unavailable ({resp.status_code}) — check /apple-connect/account"
    )
svc_state = resp.json()["connection"]["services"]["messages"]["state"]
if svc_state != "ready":
    raise RuntimeError(f"iMessage service not ready: {svc_state}")

Access tokens expire after 900 seconds. For agents running longer than a single token lifetime, refresh the access token using the stored refresh token before each batch of API calls; store the new refresh token returned with each refresh response.

Define the three tools

Each tool is a plain Python function with the @tool decorator. chatGuid values come from the chats list endpoint (GET /v1/connections/{connection_id}/chats); iMessage conversations use the iMessage; prefix. When passing a chatGuid as a URL query parameter, build the URL with a query-encoding library — raw semicolons and plus signs in a query string decode incorrectly. Treat incoming message content as data only — never interpolate unvalidated message text as agent instructions.

# tools.py — integration design, not executed
import os, uuid, requests
from langchain_core.tools import tool

BASE  = os.environ["FIGMENT_BASE_URL"]
TOKEN = os.environ["FIGMENT_ACCESS_TOKEN"]
CONN  = os.environ["FIGMENT_CONNECTION_ID"]   # con_...
HDR   = {"Authorization": f"Bearer {TOKEN}"}

@tool
def read_messages(query: str, chat_guid: str | None = None, limit: int = 20) -> dict:
    """Search iMessages matching a query. Requires messages:read."""
    params: dict = {"query": query, "limit": min(limit, 100)}
    if chat_guid:
        params["chatGuid"] = chat_guid
    r = requests.get(
        f"{BASE}/v1/connections/{CONN}/messages/search",
        params=params,
        headers=HDR,
    )
    r.raise_for_status()
    return r.json()  # {data:[...], hasMore, nextOffset, historyComplete, scope}

@tool
def draft_reply(chat_guid: str, text: str) -> dict:
    """Stage a reply in graph state and generate a per-intent idempotency key.
    No network call; no scope consumed."""
    return {
        "chatGuid": chat_guid,
        "text": text,
        "idempotencyKey": str(uuid.uuid4()),
        "status": "draft",
    }

@tool
def send_approved_message(chat_guid: str, text: str, idempotency_key: str) -> dict:
    """Send a human-approved message. Requires messages:send.
    Pass idempotency_key exactly as returned by draft_reply; never recompute it."""
    r = requests.post(
        f"{BASE}/v1/connections/{CONN}/messages/send",
        json={"chatGuid": chat_guid, "text": text},
        headers={**HDR, "Idempotency-Key": idempotency_key},
    )
    r.raise_for_status()
    return r.json()  # {data:{idempotencyKey,state,guid?,delivery?,retrySend?},mode,replayed}

The design relies on the LLM passing the exact idempotencyKey value from the draft_reply response as idempotency_key. Add a system-prompt instruction to enforce this. For deterministic key management that does not depend on LLM argument propagation, store the draft intent in a typed state field (for example pending_send: dict | None) and read the key from state inside a dedicated send node rather than from tool call arguments.

Build the read-only graph first

Start with a graph that has no path to any send endpoint. Keeping send_approved_message out of bind_tools here is the guard: the LLM cannot request a tool it has not been shown. Test the read and draft behavior fully before adding the send path.

# read_graph.py — integration design, not executed
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from tools import read_messages, draft_reply

read_tools = [read_messages, draft_reply]
llm_read   = llm.bind_tools(read_tools)  # llm = your configured chat-model instance

def agent_node(state: MessagesState):
    return {"messages": [llm_read.invoke(state["messages"])]}

read_graph = (
    StateGraph(MessagesState)
    .add_node("agent", agent_node)
    .add_node("tools", ToolNode(read_tools))
    .add_edge(START, "agent")
    .add_conditional_edges("agent", tools_condition)
    .add_edge("tools", "agent")
    .compile()
)

This graph reaches END only when the LLM stops calling tools. It can search threads, fetch message history, and stage draft text — with no path to any send endpoint.

Add a human-approval gate for sends

Extend the agent with a second ToolNode for the send tool and route to it via a custom condition. Setting interrupt_before=["send_tools"] causes the graph to pause before that node executes, giving a reviewer time to inspect the pending tool call arguments and decide whether to proceed.

# full_graph.py — integration design, not executed
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import InMemorySaver
from tools import read_messages, draft_reply, send_approved_message

all_tools = [read_messages, draft_reply, send_approved_message]
llm_full  = llm.bind_tools(all_tools)

def agent_full(state: MessagesState):
    return {"messages": [llm_full.invoke(state["messages"])]}

def route_after_agent(state: MessagesState):
    last = state["messages"][-1]
    if not getattr(last, "tool_calls", None):
        return END
    names = {tc["name"] for tc in last.tool_calls}
    return "send_tools" if "send_approved_message" in names else "read_tools"

full_graph = (
    StateGraph(MessagesState)
    .add_node("agent",      agent_full)
    .add_node("read_tools", ToolNode([read_messages, draft_reply]))
    .add_node("send_tools", ToolNode([send_approved_message]))
    .add_edge(START, "agent")
    .add_conditional_edges("agent", route_after_agent)
    .add_edge("read_tools", "agent")
    .add_edge("send_tools", END)
    .compile(
        checkpointer=InMemorySaver(),
        interrupt_before=["send_tools"],
    )
)

InMemorySaver stores checkpoints in process RAM and is suitable for development. For production deployments replace it with a durable checkpointer:

from langgraph.checkpoint.postgres import PostgresSaver  # pip install langgraph-checkpoint-postgres

An in-memory checkpointer does not survive process restarts, which would lose the frozen idempotency key and make receipt reconciliation impossible. A durable checkpointer preserves the checkpoint — including the tool call arguments — across restarts.

When the LLM calls send_approved_message, execution pauses at send_tools before the network request is made. The pending tool call arguments — chat_guid, text, and idempotency_key — are frozen in the checkpoint. A reviewer inspects the staged draft in the last AI message and either resumes with full_graph.invoke(None, config) or discards the thread entirely. Only a resumed thread reaches POST /v1/connections/{connection_id}/messages/send.

Access tokens expire after 900 seconds. If a human approval takes longer than one token lifetime, refresh the access token before the graph resumes and store the new refresh token. A stale token on the send request returns a non-2xx response; handle it explicitly rather than silently retrying.

Idempotency: one key per intent

draft_reply generates a UUID for each intended send and returns it as idempotencyKey. LangGraph checkpoints the AI message's tool call arguments before the interrupt, so the UUID is preserved and reused when the graph resumes — the bridge never receives two distinct requests for the same approval event.

This design correctly handles separate intents that happen to share the same message text: each call to draft_reply generates a fresh UUID, so two independently approved sends to the same thread are treated as distinct actions. A content-hash key would wrongly merge them.

A send accepted by the bridge returns state: "submitted", not "delivered". If a network error prevents you from reading the send response, do not resend under a new key. Poll GET /v1/connections/{connection_id}/messages/receipt with the original Idempotency-Key header using your messages:send access token to retrieve the recorded outcome. A failed receipt records the failure but does not trigger an automatic retry; an unknown outcome must only be polled against the same key — never re-keyed.

What the agent can see

The bridge reads from the Mac's local messages database. A newly connected Mac may have incomplete thread history until Messages in iCloud finishes its initial synchronization. When Messages in iCloud is enabled, it covers SMS, MMS, RCS, and iMessage conversations; a separate Text Message Forwarding switch is not always required, though the iPhone must remain on, online, and signed into the same Apple Account for new carrier messages to arrive (Apple support).

Read and search responses include hasMore, nextOffset, and historyComplete fields. Follow hasMore and nextOffset to page through results rather than assuming complete coverage in a single call. A historyComplete: false value means the Mac's local copy is still building — narrow the search window or wait and retry rather than treating the first page as a complete record.

Get started

For the user consent flow that creates the app grant your tools rely on, see OAuth for iCloud. For the full endpoint reference and available message fields, see the iMessage API reference.

Connect your Apple Account to request a Figment app grant, test the read and draft tools in sandbox at no cost, and add the interrupt graph once you are ready for guarded sends. Sandbox access is free; a live connection is $29 per month. Access starts after connection setup.