CrewAI

pip install crewai 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 crewai.tools import 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()

@tool("Check physical mailbox")
def check_mailbox() -> str:
    """Check for new postal mail at your CMRA mailing address.
    Returns carrier, status, and package IDs for each piece."""
    r = requests.get(f"{BASE}/packages?status=received", headers=HEADERS)
    packages = r.json().get("packages", [])
    if not packages:
        return "No new mail at the mailbox."
    return "\n".join(
        f"- {p.get('carrier', 'USPS')} | {p['status']} | photos: {p.get('photos_count', 0)} (ID: {p['id']})"
        for p in packages
    )


@tool("Scan a piece of mail")
def scan_mail(package_id: str) -> str:
    """Open and scan a specific piece of mail with OCR text extraction.
    Provide the package ID from check_mailbox 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']}. Results available in 1-5 minutes."
    return f"Error: {r.text}"


@tool("Get scan results")
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')}"


@tool("Forward mail to an address")
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.
    The facility re-packages and ships it."""
    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}"


@tool("Send a physical letter")
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 PDF letter via USPS. Provide a local path to the PDF.
    The facility prints, stuffs an envelope, stamps, and mails it.
    Photo proof of mailing is included."""
    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: {r.json()['outbound_mail']['id']}"
    return f"Error: {r.text}"


@tool("Shred junk mail")
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,
        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}"

from crewai import Agent, Task, Crew

# Define an agent with mail capabilities
mail_clerk = Agent(
    role="Mail Operations Clerk",
    goal="Monitor the mailbox, triage incoming mail, scan important "
         "documents, shred junk, and send outbound correspondence",
    backstory="You manage all physical mail for the organization. "
              "Government and legal mail gets scanned immediately. "
              "Marketing mail and credit card offers get shredded. "
              "Anything requiring a response is flagged.",
    tools=[check_mailbox, scan_mail, get_scan_results,
           forward_mail, send_letter, shred_mail],
    verbose=True,
)

# Define a task
daily_mail_check = Task(
    description="""Check the mailbox for new mail. For each piece:
    1. If from IRS, state tax, or any court — scan immediately
    2. If marketing, credit card offer, or "Current Resident" — shred
    3. If from an attorney — scan and report to user
    4. Everything else — report sender and type""",
    expected_output="Summary of all mail processed with actions taken",
    agent=mail_clerk,
)

# Run the crew
crew = Crew(agents=[mail_clerk], tasks=[daily_mail_check], verbose=True)
result = crew.kickoff()
print(result)

from crewai import Agent, Task, Crew

# Mail clerk handles physical mail
mail_clerk = Agent(
    role="Mail Clerk",
    goal="Process physical mail and extract key information",
    tools=[check_mailbox, scan_mail, get_scan_results, shred_mail],
    verbose=True,
)

# Legal analyst reviews scanned documents
legal_analyst = Agent(
    role="Legal Analyst",
    goal="Review scanned legal documents, identify deadlines, "
         "and draft appropriate responses",
    tools=[send_letter],  # can send response letters
    verbose=True,
)

# Tasks with handoff
triage = Task(
    description="Check mailbox. Scan any legal or government mail. "
                "Shred marketing mail. Report what was scanned.",
    expected_output="List of scanned documents with OCR text",
    agent=mail_clerk,
)

review = Task(
    description="Review the scanned documents. Extract deadlines. "
                "If a response is required, draft it and mail it certified.",
    expected_output="Summary of legal actions taken",
    agent=legal_analyst,
    context=[triage],  # receives output from triage
)

crew = Crew(agents=[mail_clerk, legal_analyst], tasks=[triage, review])
result = crew.kickoff()