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",
]

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_physical_inbound(limit: int = 25) -> str:
    """List physical letters, flats, and parcels held at the assigned PMB."""
    r = api_get("/inbound-items", params={"limit": limit})
    return json.dumps(r.json().get("inbound_items", []), indent=2)


@function_tool
def request_open_and_scan(inbound_item_id: str, expected_version: int) -> str:
    """Request the only supported physical action after member approval."""
    r = api_post(
        f"/inbound-items/{inbound_item_id}/actions",
        headers={"Idempotency-Key": f"open-scan-{inbound_item_id}-{expected_version}"},
        json={
            "action_type": "open_and_scan",
            "expected_version": expected_version,
        },
    )
    return json.dumps(r.json(), indent=2)


@function_tool
def get_physical_scan_results(inbound_item_id: str) -> str:
    """Read completed scans for one physical inbound item."""
    r = api_get(f"/inbound-items/{inbound_item_id}/scans")
    return json.dumps(r.json(), 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)

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_physical_inbound only for custody at the assigned PMB
- Request only open_and_scan, only after member approval, with the current item version
- Never infer another physical handling action
- 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 high-value, proof-sensitive, or operator-sensitive mail
- Choose certified when USPS proof of mailing and delivery is required; choose certified_return_receipt when electronic return-receipt evidence is required""",
    tools=[
        list_inbound_aliases,
        list_inbound_mail,
        get_inbound_mail,
        list_physical_inbound,
        request_open_and_scan,
        get_physical_scan_results,
        send_letter,
    ],
)

# Synchronous
result = Runner.run_sync(
    mail_agent,
    "Review forwarded inbound context, then report physical items that need member review."
)
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 digital inbound context and assigned-PMB physical mail.
Check forwarded inbound context first. For physical custody, request only
open_and_scan after member approval and include the current item version.
Hand off to Outbound Specialist when a reply letter should be drafted or mailed.""",
    tools=[list_inbound_mail, get_inbound_mail, list_physical_inbound,
           request_open_and_scan, get_physical_scan_results],
    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.