OpenAI Agents SDK

pip install openai-agents 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 agents import Agent, Runner, function_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()

@function_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."
    return "\n".join(
        f"- {p.get('carrier', 'USPS')} | {p['status']} | photos: {p.get('photos_count', 0)} (id: {p['id']})"
        for p in packages
    )


@function_tool
def scan_mail(package_id: str) -> str:
    """Request an OCR document scan for a piece of mail.
    The facility opens the envelope, scans the contents, and extracts text.
    Results available in 1-5 minutes via get_scan_results."""
    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}"


@function_tool
def get_scan_results(package_id: str) -> str:
    """Retrieve 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."
    return f"OCR text:\n{scans[0].get('ocr_text', 'N/A')}"


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


@function_tool
def send_letter(recipient_name: str, recipient_line1: str,
                recipient_city: str, recipient_state: str,
                recipient_zip: str, pdf_path: str,
                mail_class: str = "first_class") -> str:
    """Print and mail a physical letter via USPS. Provide a local PDF path.
    mail_class options: first_class, priority, certified,
    certified_return_receipt"""
    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": mail_class,
            },
        )
    if r.status_code == 201:
        return f"Letter queued. ID: {r.json()['outbound_mail']['id']}"
    return f"Error: {r.text}"


@function_tool
def shred_mail(package_id: str) -> str:
    """Shred and securely dispose of a piece of junk mail."""
    r = requests.post(
        f"{BASE}/packages/{package_id}/actions",
        headers=HEADERS,
        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}"

mail_agent = Agent(
    name="Mail Operations",
    instructions="""You manage physical mail for the organization.

Rules:
- Check the mailbox when asked
- Scan anything from IRS, state tax agencies, courts, or attorneys
- Shred credit card offers, marketing mailers, and "Current Resident" mail
- Forward legal documents to the home address when requested
- Report all new mail with sender and type
- When sending letters, always use certified mail for legal correspondence""",
    tools=[check_mailbox, scan_mail, get_scan_results,
           forward_mail, send_letter, shred_mail],
)

# Synchronous
result = Runner.run_sync(mail_agent, "Check the mailbox and handle everything.")
print(result.final_output)

# Async
import asyncio

async def main():
    result = await Runner.run(mail_agent, "Check the mailbox and scan any government mail.")
    print(result.final_output)

asyncio.run(main())

from agents import Agent, Runner

legal_reviewer = Agent(
    name="Legal Reviewer",
    instructions="""You review scanned legal documents.
Extract deadlines, identify required actions, and draft response letters.
When a response must be mailed, use send_letter with certified.""",
    tools=[get_scan_results, send_letter],
)

mail_clerk = Agent(
    name="Mail Clerk",
    instructions="""You process physical mail.
Scan government and legal mail, shred junk.
Hand off scanned legal documents to Legal Reviewer for analysis.""",
    tools=[check_mailbox, scan_mail, shred_mail],
    handoffs=[legal_reviewer],
)

# The clerk processes mail and hands off legal docs to the reviewer
result = Runner.run_sync(
    mail_clerk,
    "Check the mailbox. Scan any legal mail, then hand it to Legal Reviewer."
)
print(result.final_output)

from agents import Agent, Runner, InputGuardrail, GuardrailFunctionOutput

async def block_shred_legal(ctx, agent, input_text):
    """Prevent shredding mail that might be legal or government correspondence."""
    keywords = ["court", "legal", "attorney", "irs", "tax", "government", "subpoena"]
    if any(kw in input_text.lower() for kw in keywords):
        return GuardrailFunctionOutput(
            output_info={"reason": "Cannot shred potential legal mail"},
            tripwire_triggered=True,
        )
    return GuardrailFunctionOutput(
        output_info={"reason": "OK"},
        tripwire_triggered=False,
    )

safe_mail_agent = Agent(
    name="Safe Mail Agent",
    instructions="Process mail. Never shred government or legal mail.",
    tools=[check_mailbox, scan_mail, shred_mail, forward_mail],
    input_guardrails=[
        InputGuardrail(guardrail_function=block_shred_legal),
    ],
)