Build a support inbox with Python

Create a mailbox, send it a test message and receive an acknowledgement in the same conversation. This application records a local support case and keeps the incoming message, reply and case identifiers together.

The case record belongs to this example. The response confirms receipt; it does not resolve a support request, change an order or run a language model. Use a dedicated mailbox and messages you control.

Download the example files and guide.

Prepare the project

Use Python 3.10 or later and an API key with the Email mailbox and mailbox management permissions described in mailbox setup. The SDK selects the region from the key. Review Email pricing for mailbox and sending terms.

Contoh kode

python -m venv .venv
source .venv/bin/activate
pip install messagebird-sdk
export BIRD_API_KEY="bk_us1_..."

Save the following as mailbox_app.py in an empty project directory. The program writes mailbox-state.json next to where you run it. Keep that file private and out of source control.

Add the application

Contoh kode

import json
import os
import sys
from pathlib import Path
from uuid import uuid4
from bird import Bird

STATE = Path("mailbox-state.json")
LOCK = Path("mailbox-state.json.lock")


def run(command, first=None, second=None):
    handle = os.open(LOCK, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
    os.write(handle, str(os.getpid()).encode())
    try:
        state = json.loads(STATE.read_text()) if STATE.exists() else {
            "demoId": str(uuid4()), "cases": {}, "jobs": {}
        }

        def save():
            temporary = STATE.with_suffix(".json.tmp")
            with os.fdopen(os.open(temporary, os.O_CREAT | os.O_TRUNC | os.O_WRONLY, 0o600), "w") as stream:
                json.dump(state, stream, indent=2)
            temporary.replace(STATE)

        with Bird(api_key=os.environ["BIRD_API_KEY"], max_retries=0) as bird:
            if command == "init":
                if not state.get("mailbox"):
                    if state.get("createKey"):
                        raise RuntimeError("Creation needs review. Locate the mailbox, then use attach MAILBOX_ID.")
                    state["createKey"] = str(uuid4())
                    save()
                    mailbox = bird.email.mailboxes.create(
                        display_name="Support demo", receive_policy="open",
                        metadata={"demo_id": state["demoId"]},
                        options={"idempotency_key": state["createKey"]},
                    )
                    state["mailbox"] = {"id": mailbox.id, "address": mailbox.address}
                    save()
                print(json.dumps(state["mailbox"]))
                return
            if command == "attach" and first:
                if not state.get("createKey") or state.get("mailbox"):
                    raise RuntimeError("No unresolved mailbox creation.")
                mailbox = bird.email.mailboxes.get(first)
                if (mailbox.metadata or {}).get("demo_id") != state["demoId"]:
                    raise RuntimeError("Mailbox belongs to a different demo operation.")
                state["mailbox"] = {"id": mailbox.id, "address": mailbox.address}
                save()
                return
            if not state.get("mailbox"):
                raise RuntimeError("Run init first.")
            if command == "resolve" and first and second:
                job = state["jobs"].get(first)
                if not job or job["state"] != "uncertain":
                    raise RuntimeError("No uncertain reply for that message.")
                thread = bird.email.threads.get(job["thread"])
                reply = bird.email.threads.messages.get(job["thread"], second)
                sent = bird.email.get(second)
                metadata = sent.metadata or {}
                if (thread.mailbox_id != state["mailbox"]["id"] or reply.direction != "outbound"
                    or reply.thread_id != job["thread"] or metadata.get("demo_operation_id") != job["key"]
                    or metadata.get("demo_message_id") != first):
                    raise RuntimeError("Reply does not match the saved operation.")
                job.update(state="accepted", replyId=reply.id, replyStatus=reply.status)
                save()
                return
            if command == "status":
                print(json.dumps(state, indent=2))
                return
            if command != "sync":
                raise RuntimeError("Use init, sync [THREAD_ID], status, attach MAILBOX_ID, or resolve RECEIVED_ID SENT_ID.")

            def latest(thread_id):
                return next(iter(bird.email.threads.messages.list(
                    thread_id, direction="inbound", label="inbox", include="extracted_text", limit=1,
                )), None)

            def process_thread(thread_id):
                thread = bird.email.threads.get(thread_id)
                if thread.mailbox_id != state["mailbox"]["id"]:
                    raise RuntimeError("Thread belongs to another mailbox.")
                if "inbox" not in thread.labels:
                    return
                if any(j["thread"] == thread_id and j["state"] == "uncertain" for j in state["jobs"].values()):
                    print(json.dumps({"thread": thread_id, "result": "Review the uncertain reply before continuing."}))
                    return
                message = latest(thread_id)
                if message is None:
                    return
                job = state["jobs"].get(message.id)
                if job and job["state"] in ("accepted", "superseded"):
                    return
                if not job:
                    if len(state["jobs"]) >= 100:
                        raise RuntimeError("Demo limit reached: 100 messages. Retain state and review before continuing.")
                    attachments = bird.email.threads.messages.attachments(thread_id, message.id)
                    case_id = state["cases"].setdefault(thread_id, "case_" + str(uuid4()))
                    job = state["jobs"][message.id] = {
                        "thread": thread_id, "caseId": case_id, "key": str(uuid4()), "state": "prepared",
                        "attachments": [{"id": a.id, "filename": a.filename, "content_type": a.content_type, "size": a.size} for a in attachments.data],
                        "text": f"We recorded your message under {case_id}. Your request is awaiting review.",
                    }
                    save()
                    print(json.dumps({"case": case_id, "source": message.id, "preview": (message.extracted_text or "")[:200], "attachments": job["attachments"]}))
                current = latest(thread_id)
                if current is None or current.id != message.id:
                    job["state"] = "superseded"
                    save()
                    return
                job["state"] = "uncertain"
                save()
                reply = bird.email.threads.messages.reply(
                    thread_id, message.id, text=job["text"],
                    metadata={"demo_operation_id": job["key"], "demo_message_id": message.id, "demo_case_id": job["caseId"]},
                    options={"idempotency_key": job["key"]},
                )
                job.update(state="accepted", replyId=reply.id, replyStatus=reply.status)
                save()
                print(json.dumps({"case": job["caseId"], "message": message.id, "reply": reply.id, "status": reply.status}))

            if first:
                process_thread(first)
            else:
                for thread in bird.email.threads.list(mailbox_id=state["mailbox"]["id"], label=["inbox"]):
                    process_thread(thread.id)
    finally:
        os.close(handle)
        LOCK.unlink()


if __name__ == "__main__":
    try:
        run(*(sys.argv[1:] or ["status"]))
    except Exception as error:
        print(str(error), file=sys.stderr)
        sys.exit(1)

The application uses the newest inbox message in each thread. If several messages arrive between runs, it acknowledges the most recent one; earlier messages remain readable through the thread API. A later message updates the same local case. The demo stops after recording 100 source messages, an application limit you can change when replacing its local store.

The attachment list contains metadata. The application retains identifiers and lists filenames and sizes without downloading or executing the files. See process mailbox events and attachments to retrieve bytes deliberately.

Receive and answer a message

Contoh kode

python mailbox_app.py init

Copy the returned address and email it from a mailbox you control. Add a small PDF if you want to inspect attachment metadata. Then run:

Contoh kode

python mailbox_app.py sync
python mailbox_app.py status

You should see a case ID, a received-message ID and an outbound reply ID. A reply status of accepted means it was accepted for sending. Inspect the outgoing message in the thread reference or follow its Email events for the delivery outcome.

Run sync again. An already recorded reply is retained. Reply from your mail client once more and run sync; the new message should have a different source ID and the same case ID.

To process a single thread, pass its identifier:

Contoh kode

python mailbox_app.py sync "$THREAD_ID"

The worker checks that the thread belongs to the configured mailbox. A signed incoming event can schedule this command with its data.thread_id; follow the event handoff before accepting webhook input.

Recover an uncertain operation

The program saves the operation before making a mutation and disables automatic SDK retries. If a response is lost, it leaves the result uncertain and stops further replies for that thread. Re-running sync does not create another send.

Find the candidate outgoing message in the thread, then reconcile it:

Contoh kode

python mailbox_app.py resolve "$RECEIVED_MESSAGE_ID" "$SENT_MESSAGE_ID"

The program reads the candidate's thread and sending log, verifies the saved operation and source-message metadata, and records the returned status. A message from an unrelated operation is rejected. Sending logs have a separate retention window, so investigate before those records expire.

If mailbox creation lost its response, locate the candidate in the mailbox list and run:

Contoh kode

python mailbox_app.py attach "$MAILBOX_ID"

The mailbox must contain the demo's saved demo_id. This command performs reads and updates local state; it does not create another mailbox.

If no matching result can be established, keep the operation unresolved and investigate. Idempotency retains completed responses for a finite window and does not make a remote mutation and this local file one transaction. Do not delete the state file to bypass an uncertain result.

Prepare a deployed worker

The exclusive lock prevents two local commands from modifying the file together. After an interrupted process, confirm that its recorded PID is no longer running before removing the stale .lock file. The JSON file is a development checkpoint, not a substitute for a transactional production database or protection against a host failure.

For deployment, move cases, source-message deduplication and approved replies into your application's database. Verify and durably enqueue incoming events before acknowledging them, serialize work for each thread, and reconcile missed events within retention. Keep the acknowledgement separate from any account, order or payment action that requires application authorization.

The pre-send read detects a newer message observed before sending; it cannot prevent a message arriving immediately afterward. A consequential agent action needs its own current-state check. The fixed acknowledgement in this example makes no business-change claim.

Apply a retention and deletion policy to your local copies, including attachment metadata and development logs. Mailbox troubleshooting covers receive rules, missing messages, stale replies and delivery diagnosis.

Continue building

Support-agent workflow: add an application lookup, reviewed draft and authorized action.
Events and attachments: connect signed deliveries and inspect a file.
Choose an email integration: sending, inbound processing, hosted mailboxes and connected accounts.
Compare mailbox providers and migrate an integration.
Mailbox resources and start with Agent Mailboxes.

Sumber daya terkait

Lanjutkan dengan dokumentasi, panduan, dan contoh untuk topik ini. Sumber daya tersedia dalam bahasa Inggris.

Tonton panduannyaGetting started with email Jelajahi kemampuannyaEmail Ikuti jalur pembelajaranBuild your first integration Panduan implementasiSend your first email

Coba praktiknya dan dapatkan ringkasan implementasi