LlamaIndex

pip install llama-index llama-index-llms-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 llama_index.core.tools import FunctionTool
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()


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)


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, "X-Mailbox-MD-Version": MD_VERSION},
        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}"


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('raw_text', 'N/A')}"


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, "X-Mailbox-MD-Version": MD_VERSION},
        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}"


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}"


def shred_mail(package_id: str) -> str:
    """Shred and securely dispose of a piece of mail."""
    r = requests.post(
        f"{BASE}/packages/{package_id}/actions",
        headers={**HEADERS, "X-Mailbox-MD-Version": MD_VERSION},
        json={"action_type": "shred"},
    )
    if r.status_code == 201:
        return f"Shred requested. Action ID: {r.json()['action_request']['id']}"
    return f"Error: {r.text}"


# Wrap each function as a LlamaIndex tool
check_mailbox_tool = FunctionTool.from_defaults(fn=check_mailbox)
scan_mail_tool = FunctionTool.from_defaults(fn=scan_mail)
get_scan_results_tool = FunctionTool.from_defaults(fn=get_scan_results)
forward_mail_tool = FunctionTool.from_defaults(fn=forward_mail)
send_letter_tool = FunctionTool.from_defaults(fn=send_letter)
shred_mail_tool = FunctionTool.from_defaults(fn=shred_mail)

from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI

tools = [
    check_mailbox_tool, scan_mail_tool, get_scan_results_tool,
    forward_mail_tool, send_letter_tool, shred_mail_tool,
]

llm = OpenAI(model="gpt-4o")
agent = ReActAgent.from_tools(
    tools,
    llm=llm,
    system_prompt="""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. Shred junk mail. Report all new mail with sender and type.""",
    verbose=True,
)

# Run it
response = agent.chat("Check the mailbox and scan any government mail.")
print(response)

from llama_index.core.agent.function_calling import FunctionCallingAgent

agent = FunctionCallingAgent.from_tools(
    tools,
    llm=llm,
    system_prompt="You manage physical mail for the organization.",
    verbose=True,
)

response = agent.chat("Check for new mail and shred anything from advertisers.")

# Flask webhook handler example
from flask import Flask, request, jsonify
import hmac, hashlib, 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"] == "package.received":
        agent.chat(
            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})