OpenAI Agents SDK

pip install openai-agents 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

import json
from typing import Literal

from agents import Agent, Runner, function_tool
import requests

session = requests.Session()
session.headers.update(HEADERS)

MailClass = Literal[
    "first_class",
    "priority",
    "certified",
    "certified_return_receipt",
    "fedex_ground",
    "fedex_express",
    "fedex_2day",
    "fedex_overnight",
    "ups_ground",
    "ups_2day",
    "ups_next_day",
]

ScanType = Literal["label", "envelope", "document", "content"]


def api_get(path: str, **kwargs):
    r = session.get(f"{BASE}{path}", timeout=30, **kwargs)
    r.raise_for_status()
    return r


def api_post(path: str, *, headers: dict | None = None, **kwargs):
    merged_headers = dict(HEADERS)
    if headers:
        merged_headers.update(headers)
    r = session.post(f"{BASE}{path}", headers=merged_headers, timeout=60, **kwargs)
    r.raise_for_status()
    return r


def get_mailbox_md_version() -> str:
    r = api_get(f"/agents/{AGENT_ID}/instructions")
    return str(r.json()["version"])


def mailbox_md_headers() -> dict:
    return {"X-Mailbox-MD-Version": get_mailbox_md_version()}


@function_tool
def list_inbound_aliases() -> str:
    """List the private forwarding aliases you can send scans, PDFs, photos, and notes to."""
    r = api_get("/inbound-forwarding-addresses")
    aliases = r.json().get("forwarding_addresses", [])
    if not aliases:
        return "No forwarding aliases found."
    return json.dumps(
        [
            {
                "id": a["id"],
                "label": a.get("label"),
                "email": a.get("email"),
                "source_type": a.get("source_type"),
                "provider": a.get("provider"),
            }
            for a in aliases
        ],
        indent=2,
    )


@function_tool
def list_inbound_mail(limit: int = 10, include_drafting: bool = True) -> str:
    """List forwarded inbound mail context from existing addresses.
    include_drafting adds reply-ready context for outbound follow-ups."""
    include = "drafting" if include_drafting else ""
    params = {"limit": limit}
    if include:
        params["include"] = include
    r = api_get("/inbound", params=params)
    items = r.json().get("inbound_mail", [])
    if not items:
        return "No inbound mail context found."
    return json.dumps(items, indent=2)


@function_tool
def get_inbound_mail(inbound_id: str, include_files: bool = True) -> str:
    """Get one inbound mail item with summary, drafting context, and optional source files."""
    include = ["drafting", "lineage"]
    if include_files:
        include.append("files")
    r = api_get(
        f"/inbound/{inbound_id}",
        params={"include": ",".join(include)},
    )
    return json.dumps(r.json()["inbound_mail"], indent=2)


@function_tool
def list_packages(status: str = "received") -> str:
    """List physical mail and packages already received at the mailbox."""
    r = api_get("/packages", params={"status": status})
    packages = r.json().get("packages", [])
    if not packages:
        return "No new mail."
    return json.dumps(packages, indent=2)


@function_tool
def request_scan(package_id: str, scan_type: ScanType = "content") -> str:
    """Request a scan for a package.
    Use content to open and scan contents, document for document OCR, label for labels, envelope for exterior."""
    r = api_post(
        f"/packages/{package_id}/scan",
        json={"scan_type": scan_type},
    )
    scan = r.json()["scan"]
    return json.dumps(
        {
            "scan_id": scan["id"],
            "package_id": scan["package_id"],
            "scan_type": scan["scan_type"],
            "status": scan["status"],
            "action_request_id": scan.get("action_request_id"),
        },
        indent=2,
    )


@function_tool
def get_scan_results(package_id: str) -> str:
    """Retrieve completed scan rows for a previously scanned package."""
    r = api_get(f"/packages/{package_id}/scan")
    scans = r.json().get("scans", [])
    if not scans:
        return "No scan results yet. The scan may still be processing."
    return json.dumps(scans, indent=2)


@function_tool
def forward_package(
    package_id: str,
    to_name: str,
    to_line1: str,
    to_city: str,
    to_state: str,
    to_zip: str,
    to_line2: str = "",
    to_country: str = "US",
    service_level: str = "ground",
    carrier_preference: str = "",
    insurance: bool = False,
    signature_required: bool = False,
) -> str:
    """Forward a received package or piece of mail to another address."""
    r = api_post(
        f"/packages/{package_id}/forward",
        headers=mailbox_md_headers(),
        json={
            "to_name": to_name,
            "to_line1": to_line1,
            "to_line2": to_line2 or None,
            "to_city": to_city,
            "to_state": to_state,
            "to_zip": to_zip,
            "to_country": to_country,
            "service_level": service_level,
            "carrier_preference": carrier_preference or None,
            "insurance": insurance,
            "signature_required": signature_required,
        },
    ).json()["forwarding_request"]
    return json.dumps(r, indent=2)


@function_tool
def send_letter(
    recipient_name: str,
    recipient_line1: str,
    recipient_city: str,
    recipient_state: str,
    recipient_zip: str,
    pdf_path: str,
    recipient_line2: str = "",
    recipient_country: str = "US",
    mail_class: MailClass = "first_class",
    return_name: str = "",
    return_company: str = "",
    return_line1: str = "",
    return_line2: str = "",
    return_city: str = "",
    return_state: str = "",
    return_zip: str = "",
    agent_notes: str = "",
    inbound_capture_id: str = "",
    postal_mail_thread_id: str = "",
    requires_approval: bool = False,
    dry_run: bool = False,
) -> str:
    """Print and mail a physical letter. Supports USPS, FedEx, and UPS classes.
    Use dry_run=True to preview cost without creating a record or charging."""
    with open(pdf_path, "rb") as f:
        r = api_post(
            "/mail",
            headers=mailbox_md_headers(),
            files={"document": ("letter.pdf", f, "application/pdf")},
            data={
                "recipient_name": recipient_name,
                "recipient_line1": recipient_line1,
                "recipient_line2": recipient_line2,
                "recipient_city": recipient_city,
                "recipient_state": recipient_state,
                "recipient_zip": recipient_zip,
                "recipient_country": recipient_country,
                "mail_class": mail_class,
                "return_name": return_name,
                "return_company": return_company,
                "return_line1": return_line1,
                "return_line2": return_line2,
                "return_city": return_city,
                "return_state": return_state,
                "return_zip": return_zip,
                "agent_notes": agent_notes,
                "requires_approval": str(requires_approval).lower(),
                "dry_run": str(dry_run).lower(),
                "inbound_capture_id": inbound_capture_id,
                "postal_mail_thread_id": postal_mail_thread_id,
            },
        )
    payload = r.json()
    if payload.get("dry_run"):
        return json.dumps(payload, indent=2)
    return json.dumps(payload["outbound_mail"], indent=2)


@function_tool
def shred_mail(package_id: str) -> str:
    """Shred and securely dispose of a piece of junk mail or spam."""
    r = api_post(
        f"/packages/{package_id}/actions",
        headers=mailbox_md_headers(),
        json={"action_type": "shred"},
    )
    return json.dumps(r.json()["action_request"], indent=2)

mail_agent = Agent(
    name="Mail Ops",
    instructions="""You manage inbound and outbound postal workflows.

Rules:
- Start with inbound context from forwarded aliases when the user mentions PDFs, scans, notices, or email-forwarded mail
- Use list_packages for physically received items already at the mailbox
- Use request_scan with scan_type="content" when the contents need to be opened and OCR'd
- Use draft_context from inbound mail when preparing a reply letter tied to a prior inbound item
- Use dry_run=True before expensive or sensitive outbound mail
- Use requires_approval=True for legal, financial, or operator-sensitive mail
- Prefer certified or certified_return_receipt for legal correspondence""",
    tools=[
        list_inbound_aliases,
        list_inbound_mail,
        get_inbound_mail,
        list_packages,
        request_scan,
        get_scan_results,
        forward_package,
        send_letter,
        shred_mail,
    ],
)

# Synchronous
result = Runner.run_sync(
    mail_agent,
    "Review new inbound items, then check received packages and flag anything urgent."
)
print(result.final_output)

# Async
import asyncio

async def main():
    result = await Runner.run(
        mail_agent,
        "Check forwarded inbound context and prepare a reply for any time-sensitive notice."
    )
    print(result.final_output)

asyncio.run(main())

from agents import Agent, Runner

outbound_specialist = Agent(
    name="Outbound Specialist",
    instructions="""You prepare outbound letters.
Use draft_context from inbound items when available.
Run dry_run=True first for sensitive or expensive mail, then send with
requires_approval=True when a human should confirm before fulfillment.""",
    tools=[get_inbound_mail, send_letter],
)

inbound_triage = Agent(
    name="Inbound Triage",
    instructions="""You review inbound context and physical mail.
Check forwarded inbound items first, then packages if needed.
Hand off to Outbound Specialist when a reply letter should be drafted or mailed.""",
    tools=[list_inbound_mail, get_inbound_mail, list_packages, request_scan, get_scan_results, shred_mail],
    handoffs=[outbound_specialist],
)

result = Runner.run_sync(
    inbound_triage,
    "Review new inbound notices and mail a reply if one is required."
)
print(result.final_output)

preview = send_letter(
    recipient_name="County Clerk",
    recipient_line1="123 Main St",
    recipient_city="Sacramento",
    recipient_state="CA",
    recipient_zip="95814",
    pdf_path="/tmp/reply.pdf",
    mail_class="certified",
    inbound_capture_id="00000000-0000-0000-0000-000000000000",
    requires_approval=True,
    dry_run=True,
)

# preview contains cost_breakdown, warnings, and normalized mail settings.
# After review, send again with dry_run=False to create the outbound record.