# 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](/downloads/guide-mailboxes-python-sample.zip). ## 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](/docs/get-started/agent-mailboxes). The SDK selects the region from the key. Review [Email pricing](/products/email/pricing) for mailbox and sending terms. ```bash 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 ```python 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](/docs/guides/mailboxes/events-and-attachments) to retrieve bytes deliberately. ## Receive and answer a message ```bash 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: ```bash 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](/docs/api/reference/get-email-thread-message) or follow its [Email events](/docs/guides/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: ```bash 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](/docs/guides/mailboxes/events-and-attachments) 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: ```bash 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](/docs/guides/email/mailboxes#retention-and-erasure), so investigate before those records expire. If mailbox creation lost its response, locate the candidate in the [mailbox list](/docs/api/reference/list-mailboxes) and run: ```bash 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](/docs/guides/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](/docs/guides/mailboxes/troubleshooting) covers receive rules, missing messages, stale replies and delivery diagnosis. ## Continue building - [Support-agent workflow](/docs/guides/mailboxes/application-flow): add an application lookup, reviewed draft and authorized action. - [Events and attachments](/docs/guides/mailboxes/events-and-attachments): connect signed deliveries and inspect a file. - [Choose an email integration](/docs/guides/mailboxes/choose-integration): sending, inbound processing, hosted mailboxes and connected accounts. - [Compare mailbox providers](/products/mailboxes/compare) and [migrate an integration](/docs/guides/mailboxes/migrate). - [Mailbox resources](/mailboxes/resources) and [start with Agent Mailboxes](/agent-mailboxes). ## Related resources - [Getting started with email](/learn/email/getting-started-with-email) (video) - [Email](/email-api) (product) - [Build your first integration](/learn/paths/integration) (course) - [Send your first email](/docs/get-started/send-your-first-email) (docs) [Get an implementation brief](/learn/workspace?topic=email)