Skip to content
Linkworld Docs

Reacting to inbound emails

If your app needs to do something when an email arrives — classify a support ticket, match a reply to a sales sequence, file an invoice — don’t poll email_search. The platform already turns every inbound email into a normalized event and delivers it to your on_inbound handler, with the thread context resolved.

The mental shift. Email isn’t a thing you fetch. It’s an event the platform pushes. Your code is a reaction, not a checker.

What you get

Section titled “What you get”

When a tenant receives an email and your app is wired up, the platform calls:

@app.on_inbound
async def handler(ctx: Context, env: InboundEnvelope) -> None:
    if env.channel != "email":
        return  # ignore non-email inbounds


    # env.channel_thread_id   → already-resolved thread, stable across replies
    # env.channel_message_id  → this specific message's id (for email_reply)
    # env.sender_name         → 'Alex Smith'
    # env.user_id             → tenant user the mail is for
    # env.message_text        → plain-text body
    # env.attachments         → [{file_id, filename, content_type}, …]


    await ctx.tools.call(
        "email_reply",
        message_id=env.channel_message_id,
        body="Thanks — I'll get back to you.",
    )

The same envelope shape is used for WhatsApp, web chat, voice transcripts, and webhook — switching channels means switching the env.channel != "email" guard. See InboundEnvelope for all fields.

Canonical solution

Section titled “Canonical solution”

1. Enable inbound in your manifest

Section titled “1. Enable inbound in your manifest”
linkworld.app.yaml
lifecycle:
  on_inbound: true       # this app implements an on_inbound handler


required_scopes:
  - mail.read            # to read message + thread context
  - mail.send            # if you plan to reply

Setting lifecycle.on_inbound: true tells the platform your app implements the handler. The tenant then opts in to fan-out through their workspace-control UI — without that opt-in, the platform routes inbound messages to its standard agent and your app never sees them. There’s no subscribes_to: entry for inbound channels (that field is for cross-app events).

2. Filter by channel and route

Section titled “2. Filter by channel and route”
@app.on_inbound
async def on_inbound(ctx, env):
    if env.channel != "email":
        return


    # App-private records live in ctx.kv under `{record_type}:{id}`
    # keys. List by prefix, then filter in code on the lookup field.
    items = await ctx.kv.list(
        prefix="touch:", limit=200, include_values=True,
    )
    match = next(
        (item for item in items
         if isinstance(item.get("value"), dict)
         and item["value"].get("email_thread_id") == env.channel_thread_id),
        None,
    )


    if match:
        await handle_campaign_reply(ctx, env, match["value"])
    else:
        await handle_new_inquiry(ctx, env)

Persistence note. App-private records (touches, drafts, suppression rows, sequence templates) live in ctx.kv for marketplace apps. Keys follow a {record_type}:{id} convention so ctx.kv.list(prefix=...) reads back all records of a type. ctx.kv has no native indexes — filter in code after the list call. For hot lookups, denormalize the key (e.g. thread:<channel_thread_id> mapping to the touch id) or batch the match in your heartbeat instead of on_inbound.

3. Replying — use env.channel_message_id

Section titled “3. Replying — use env.channel_message_id”

The platform’s email_reply tool wants a message_id — the identifier for the specific inbound message you’re answering, not the thread. InboundEnvelope.channel_message_id carries exactly that (the M365 / Graph message_id for email inbounds), so the reply is a single tool call:

async def reply_in_thread(ctx, env, body: str):
    if env.channel_message_id is None:
        # Defensive fallback. Email inbounds always populate it,
        # but other channels (or future ones) may not. Fall back
        # to a fresh send.
        await ctx.tools.call(
            "email_send",
            to=[derive_recipient(env)],
            subject="Re: …",
            body=body,
        )
        return
    await ctx.tools.call(
        "email_reply",
        message_id=env.channel_message_id,
        body=body,
    )

channel_message_id is distinct from channel_thread_id:

  • channel_thread_id is stable across the whole conversation — use it to match an inbound to your app’s prior outbound (step 2 above).
  • channel_message_id identifies this inbound message — pass it to email_reply to thread the reply correctly.

4. Store the thread_id when sending outbound, too

Section titled “4. Store the thread_id when sending outbound, too”

For your reply-matching to work, you must record the thread_id when you sent the original email:

import uuid


# When sending a touch
result = await ctx.tools.call(
    "email_send",
    to=[lead["email"]],          # NB: array, even for a single recipient
    subject=touch_subject,
    body=personalized_body,
)
# result["conversation_id"] is the same channel_thread_id that future
# inbound replies will carry — store it so step 2 can match.
touch_id = str(uuid.uuid4())
await ctx.kv.set(f"touch:{touch_id}", {
    "id": touch_id,
    "campaign_id": campaign_id,
    "lead_id": lead["id"],
    "email_thread_id": result.get("conversation_id"),
    "touch_number": n,
})

❌ Anti-patterns

Section titled “❌ Anti-patterns”

Don’t: poll email_search to find new mail

Section titled “Don’t: poll email_search to find new mail”
# ❌ DON'T do this
async def poll_inbox(ctx):
    cursor = await ctx.kv.get("last_check") or "1970-01-01"
    new_mail = await ctx.tools.call("email_search", since=cursor)
    for mail in new_mail:
        await process(mail)
    await ctx.kv.set("last_check", now())

Why it’s wrong:

  • You re-fetch mail the platform already pushed to other handlers. Now there are two write-paths for the same data and they can diverge.
  • Latency: depends on poll interval, typically minutes. on_inbound fires within seconds of the mail arriving.
  • Pagination, cursor management, retry semantics — all your responsibility, all easy to get wrong.
  • Tenant rate-limit budget for email_search is consumed by polling instead of actual lookups.
  • If the user uninstalls and reinstalls your app, the poll cursor is lost or stale; on_inbound is automatically re-wired.

Don’t: parse In-Reply-To / References / Message-ID headers yourself

Section titled “Don’t: parse In-Reply-To / References / Message-ID headers yourself”
# ❌ DON'T do this
def find_thread_for_reply(message):
    in_reply_to = message.headers.get("In-Reply-To")
    refs = message.headers.get("References", "").split()
    # … 100 lines of M365 / Gmail / quirks-of-each-provider parsing

Why it’s wrong:

  • Every email provider formats these headers differently.
  • Forwards, replies-from-different-clients, and mailing-list rewrites break naïve parsing.
  • The platform already resolves both pieces for you: stable channel_thread_id for matching to your prior outbound, and channel_message_id for passing into email_reply. Your matching code is one == comparison; your reply is one tool call.

linkworld lint rule LWP004 flags this anti-pattern automatically.

Don’t: use email_send as a substitute for email_reply

Section titled “Don’t: use email_send as a substitute for email_reply”
# ❌ DON'T do this (unless email_search returned no message)
await ctx.tools.call(
    "email_send",
    to=[sender_email],
    subject=f"Re: {subject}",
    body=reply,
)

This creates a new thread from the recipient’s perspective, even if the subject starts with “Re:”. Resolve the message_id via email_search and use email_reply — the platform threads it correctly so the user sees one conversation in their mailbox. Only fall back to email_send if the search returned no message (rare race during initial mailbox sync).

Don’t: rely on a require_approval parameter

Section titled “Don’t: rely on a require_approval parameter”

Neither email_send nor email_reply has a require_approval parameter today. The expectation is encoded in the agent’s system prompt (“Always ask the user for confirmation before sending a reply”) and surfaced via your app’s own approval UI. For sensitive outbound, draft the message into your app’s records as status='draft_for_approval' and only call email_reply after the user confirms.

When this doesn’t apply

Section titled “When this doesn’t apply”
  • You want events from another app’s emissions, not the inbox. That’s cross-app events — on_event with a wire from the emitter app, not from inbound.email.
  • You need historical mail, not live inbound. email_search is still the right tool for “show me invoices from last quarter” — it’s the polling pattern for ad-hoc retrieval that’s wrong, not the tool itself.
  • You’re processing email at extreme rate. on_inbound is per-message; if a tenant receives 10k mails/hour, you’ll want a heartbeat with a query pre-filter that batches them, not a handler invocation per mail.
Section titled “Related”