LangChain

pip install langchain langchain-openai requests

import os

API_KEY = os.environ["MAILBOX_BOT_API_KEY"]  # sk_agent_live_*
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 check_mailbox() -> str:
    """Check for new physical mail at your postal mailing address.
    Returns a list of received mail with carrier, status, and package ID."""
    r = requests.get(f"{BASE}/packages?status=received", headers=HEADERS)
    packages = r.json().get("packages", [])
    if not packages:
        return "No new mail."
    lines = []
    for p in packages:
        carrier = p.get("carrier") or "USPS"
        lines.append(
            f"- {carrier} | {p['status']} | "
            f"photos: {p.get('photos_count', 0)} (id: {p['id']})"
        )
    return "\n".join(lines)


@tool
def scan_mail(package_id: str) -> str:
    """Request a document scan with OCR extraction for a piece of mail.
    Returns the action ID to track progress."""
    r = requests.post(
        f"{BASE}/packages/{package_id}/actions",
        headers=HEADERS,
        json={"action_type": "open_and_scan"},
    )
    if r.status_code == 201:
        return f"Scan requested. Action ID: {r.json()['action_request']['id']}"
    return f"Error: {r.text}"


@tool
def get_scan_results(package_id: str) -> str:
    """Get OCR scan results for a previously scanned piece of mail."""
    r = requests.get(
        f"{BASE}/packages/{package_id}/scan",
        headers=HEADERS,
    )
    scans = r.json().get("scans", [])
    if not scans:
        return "No scan results yet. The scan may still be processing."
    latest = scans[0]
    return f"OCR text:\n{latest.get('ocr_text', 'N/A')}"


@tool
def forward_mail(package_id: str, name: str, address: str,
                 city: str, state: str, zip_code: str) -> str:
    """Forward a piece of mail to another US address."""
    r = requests.post(
        f"{BASE}/packages/{package_id}/actions",
        headers=HEADERS,
        json={
            "action_type": "forward",
            "parameters": {
                "forward_name": name,
                "forward_address": address,
                "forward_city": city,
                "forward_state": state,
                "forward_zip": zip_code,
            },
        },
    )
    if r.status_code == 201:
        return f"Forward requested. Action ID: {r.json()['action_request']['id']}"
    return 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 langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

tools = [check_mailbox, scan_mail, get_scan_results, forward_mail, send_letter]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You manage physical mail for the organization.
Check the mailbox when asked. Scan anything from government agencies,
courts, or the IRS immediately. Forward legal documents to the home
address. Report all new mail with sender and type."""),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

llm = ChatOpenAI(model="gpt-4o")
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Run it
result = executor.invoke({"input": "Check the mailbox and scan any government mail."})

# 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"] == "package.received":
        executor.invoke({
            "input": f"New mail arrived. Package ID: {event['package_id']}. "
                     f"Carrier: {event.get('carrier', 'USPS')}. "
                     f"Decide what to do with it."
        })

    return jsonify({"ok": True})