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 list_physical_inbound() -> str:
    """List letters, flats, and parcels held at the assigned PMB."""
    r = requests.get(f"{BASE}/inbound-items", headers=HEADERS)
    return str(r.json()) if r.ok else f"Error: {r.text}"


def request_open_and_scan(inbound_item_id: str, expected_version: int) -> str:
    """Request the only supported physical action after member approval."""
    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}"


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


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



# Wrap each function as a LlamaIndex tool
list_physical_inbound_tool = FunctionTool.from_defaults(fn=list_physical_inbound)
request_open_and_scan_tool = FunctionTool.from_defaults(fn=request_open_and_scan)
get_scan_results_tool = FunctionTool.from_defaults(fn=get_scan_results)
send_letter_tool = FunctionTool.from_defaults(fn=send_letter)

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

tools = [
    list_physical_inbound_tool, request_open_and_scan_tool,
    get_scan_results_tool, send_letter_tool,
]

llm = OpenAI(model="gpt-4o")
agent = ReActAgent.from_tools(
    tools,
    llm=llm,
system_prompt="""Keep forwarded digital/OCR context separate from physical custody.
For assigned-PMB items, request only open_and_scan, only after member approval,
and always include the current item version. Infer no other physical action.""",
    verbose=True,
)

# Run it
response = agent.chat("List physical inbound items and ask before requesting Open & Scan.")
print(response)

from llama_index.core.agent.function_calling import FunctionCallingAgent

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

response = agent.chat("Check for physical inbound items and report what needs review.")

# 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"] == "inbound.received":
        agent.chat(
            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})