OpenAI Agents SDK Integration Guide

curl -X POST https://api.molted.email/v1/me/mailboxes \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Outbound Agent",
    "emailAddress": "agent@yourdomain.com"
  }'

import os
import requests
from agents import function_tool

MOLTED_API_KEY = os.environ["MOLTED_API_KEY"]
MOLTED_TENANT_ID = os.environ["MOLTED_TENANT_ID"]
MOLTED_MAILBOX_ID = os.environ["MOLTED_MAILBOX_ID"]


@function_tool
def send_email(
    to: str,
    subject: str,
    body: str,
    dedupe_key: str | None = None,
) -> str:
    """Send an email to a contact through the governed mailbox.

    The email is evaluated against policy rules before delivery.
    Blocked sends return a reason and are not retried.

    Args:
        to: Recipient email address.
        subject: Email subject line.
        body: Email body in plain text or HTML.
        dedupe_key: Optional unique key to prevent duplicate sends.
            Defaults to recipient+subject if omitted.
    """
    payload = {
        "tenantId": MOLTED_TENANT_ID,
        "mailboxId": MOLTED_MAILBOX_ID,
        "recipientEmail": to,
        "templateId": "_default",
        "dedupeKey": dedupe_key or f"{to}-{subject}",
        "agentId": "openai-agent",
        "payload": {
            "subject": subject,
            "html": body,
            "text": body,
        },
    }

    resp = requests.post(
        "https://api.molted.email/v1/agent/send",
        json=payload,
        headers={"Authorization": f"Bearer {MOLTED_API_KEY}"},
        timeout=10,
    )
    resp.raise_for_status()
    result = resp.json()

    if result["status"] == "blocked":
        return (
            f"Email blocked by policy: {result['blockReason']}. "
            f"Decision trace: {result['requestId']}"
        )

    return f"Email queued. requestId={result['requestId']}, status={result['status']}"

import os
import requests
from agents import function_tool

MOLTED_API_KEY = os.environ["MOLTED_API_KEY"]


@function_tool
def check_email_status(request_id: str) -> str:
    """Check the delivery status of a previously queued email.

    Args:
        request_id: The requestId returned when the email was sent.
    """
    resp = requests.get(
        f"https://api.molted.email/v1/agent/send/{request_id}/status",
        headers={"Authorization": f"Bearer {MOLTED_API_KEY}"},
        timeout=10,
    )
    resp.raise_for_status()
    data = resp.json()

    provider = data.get("provider", "unknown")
    last_event = data.get("lastEvent", "none")
    return f"Status: {data['status']}. Provider: {provider}. Last event: {last_event}."

import asyncio
from agents import Agent, Runner
from tools.send_email import send_email
from tools.check_email_status import check_email_status

outreach_agent = Agent(
    name="Outreach Agent",
    instructions="""
    You are a customer outreach agent. When sending emails:
    - Always supply a dedupe_key to prevent accidental duplicate sends.
    - If a send is blocked by policy, report the reason to the user - do not retry.
    - After sending, check delivery status if the user asks.
    - 'duplicate_send' means the email was already sent recently - inform the user and stop.
    - 'suppressed_recipient' means this contact cannot be emailed - inform the user and stop.
    - 'rate_limit_exceeded' means the send rate has been reached - wait and inform the user.
    """,
    tools=[send_email, check_email_status],
)


async def main():
    result = await Runner.run(
        outreach_agent,
        input=(
            "Send a trial welcome email to alice@example.com. "
            "Subject: 'Welcome to the trial' "
            "Body: 'Thanks for signing up - your 14-day trial starts now.'"
        ),
    )
    print(result.final_output)


if __name__ == "__main__":
    asyncio.run(main())

from agents import Agent, Runner
from tools.send_email import send_email
from tools.check_email_status import check_email_status

# Specialized email agent - the only one with mailbox access
email_agent = Agent(
    name="Email Agent",
    instructions="""
    You send and track emails for the team. Always use a dedupe_key.
    Report policy blocks clearly - do not retry terminal blocks.
    """,
    tools=[send_email, check_email_status],
)

# Triage agent - routes tasks and delegates email sends
triage_agent = Agent(
    name="Triage Agent",
    instructions="""
    You coordinate tasks. For any email sending, hand off to the Email Agent.
    Do not attempt to send emails directly.
    """,
    handoffs=[email_agent],
)

molted listen --pipe "python reply_handler.py" --events "inbound.*"

import sys
import json
import asyncio
from agents import Agent, Runner
from tools.send_email import send_email

reply_agent = Agent(
    name="Reply Agent",
    instructions="You handle inbound replies. Respond helpfully and concisely.",
    tools=[send_email],
)


async def handle_event(event: dict) -> None:
    if event.get("event") != "inbound.classified":
        return

    data = event["data"]
    intent = data.get("intent")
    suggested = data.get("suggestedAction")

    if intent == "support" and suggested == "reply":
        from_email = data["fromEmail"]
        subject = data["subject"]

        await Runner.run(
            reply_agent,
            input=(
                f"Reply to {from_email} who wrote: '{subject}'. "
                "Keep it short and helpful."
            ),
        )


for line in sys.stdin:
    event = json.loads(line.strip())
    asyncio.run(handle_event(event))