LangChain

pip install langchain langgraph langchain-openai requests

import os

API_KEY = os.environ["MAILBOX_BOT_API_KEY"]  # sk_agent_*
BASE = "https://mailbox.bot/api/v1"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}
AGENT_ID = os.environ["MAILBOX_BOT_AGENT_ID"]  # from dashboard

from langchain.tools import tool
import requests

# Fetch MAILBOX.md version (required for outbound mail)
def get_md_version() -> str:
    r = requests.get(f"{BASE}/agents/{AGENT_ID}/instructions", headers=HEADERS)
    return str(r.json()["version"])

MD_VERSION = get_md_version()

@tool
def list_physical_inbound() -> str:
    """List physical letters, flats, and parcels at the assigned PMB."""
    r = requests.get(f"{BASE}/inbound-items", headers=HEADERS)
    if not r.ok:
        return f"Error: {r.text}"
    return str(r.json().get("inbound_items", []))


@tool
def request_open_and_scan(inbound_item_id: str, expected_version: int) -> str:
    """Request the only supported physical action: Open & Scan."""
    r = requests.post(
        f"{BASE}/inbound-items/{inbound_item_id}/actions",
        headers={**HEADERS, "Idempotency-Key": f"open-scan-{inbound_item_id}-{expected_version}"},
        json={"action_type": "open_and_scan", "expected_version": expected_version},
    )
    return str(r.json()) if r.ok else f"Error: {r.text}"


@tool
def get_scan_results(inbound_item_id: str) -> str:
    """Read completed scans for one physical inbound item."""
    r = requests.get(
        f"{BASE}/inbound-items/{inbound_item_id}/scans",
        headers=HEADERS,
    )
    return str(r.json()) if r.ok else f"Error: {r.text}"

@tool
def send_letter(recipient_name: str, recipient_line1: str,
                recipient_city: str, recipient_state: str,
                recipient_zip: str, pdf_path: str) -> str:
    """Print and mail a physical letter. Provide a local PDF path.
    The facility prints, stuffs, stamps, and mails it."""
    with open(pdf_path, "rb") as f:
        r = requests.post(
            f"{BASE}/mail",
            headers={**HEADERS, "X-Mailbox-MD-Version": MD_VERSION},
            files={"document": ("letter.pdf", f, "application/pdf")},
            data={
                "recipient_name": recipient_name,
                "recipient_line1": recipient_line1,
                "recipient_city": recipient_city,
                "recipient_state": recipient_state,
                "recipient_zip": recipient_zip,
                "mail_class": "first_class",
            },
        )
    if r.status_code == 201:
        return f"Letter queued for mailing. ID: {r.json()['outbound_mail']['id']}"
    return f"Error: {r.text}"

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(model="gpt-4o")

agent = create_react_agent(
    llm,
    tools=[list_physical_inbound, request_open_and_scan, get_scan_results, send_letter],
    prompt="You manage postal mail for the organization.\n"
           "Use /inbound for forwarded digital/OCR context. For physical custody,\n"
           "list assigned-PMB items and request only open_and_scan with the current\n"
           "item version. Never infer another physical handling action.",
)

# Run it
result = agent.invoke(
    {"messages": [("user", "List physical inbound items and ask before requesting Open & Scan.")]}
)

# Flask webhook handler example
from flask import Flask, request, jsonify
import hmac, hashlib, time, os

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["MAILBOX_BOT_WEBHOOK_SECRET"]

@app.route("/webhook/mail", methods=["POST"])
def handle_mail_webhook():
    sig_header = request.headers.get("X-Mailbox-Signature", "")
    body = request.get_data(as_text=True)

    # Signature format: whsk_prefix:t=<timestamp>,v1=<hmac-hex>
    _, _, sig_data = sig_header.partition(":")
    params = dict(p.split("=", 1) for p in sig_data.split(",") if "=" in p)
    ts, v1 = params.get("t", ""), params.get("v1", "")

    expected = hmac.new(
        WEBHOOK_SECRET.encode(), f"{ts}.{body}".encode(), hashlib.sha256
    ).hexdigest()

    if not v1 or not hmac.compare_digest(v1, expected):
        return "invalid signature", 401

    event = request.json
    if event["event_type"] == "inbound.received":
        agent.invoke({
            "messages": [{
                "role": "user",
                "content": f"Physical mail arrived. Inbound item ID: {event['inbound_item_id']}. "
                           "Report it; request only Open & Scan when the member approves."
            }]
        })

    return jsonify({"ok": True})