AgentSend + OpenAI Agents: Email for OpenAI Agent SDK

pip install openai-agents requests

import os
import requests
from typing import Optional
from agents import function_tool

AGENTSEND_API_KEY = os.environ["AGENTSEND_API_KEY"]
AGENTSEND_BASE = "https://agentsend.io"


@function_tool
def send_email(
    to: str,
    subject: str,
    body: str,
    inbox_id: str,
    thread_id: Optional[str] = None,
) -> str:
    """Send an email via AgentSend.

    Args:
        to: Recipient email address.
        subject: Email subject line.
        body: Plain text email body.
        inbox_id: AgentSend inbox ID to send from.
        thread_id: Thread ID for replies (omit for new conversations).
    """
    payload = {
        "inboxId": inbox_id,
        "to": to,
        "subject": subject,
        "text": body,
    }
    if thread_id:
        payload["threadId"] = thread_id

    resp = requests.post(
        f"{AGENTSEND_BASE}/messages",
        json=payload,
        headers={"Authorization": f"Bearer {AGENTSEND_API_KEY}"},
    )
    resp.raise_for_status()
    data = resp.json()
    return f"Email sent. Message ID: {data['id']}, Thread: {data['threadId']}"


@function_tool
def check_inbox(inbox_id: str) -> str:
    """Check for new messages in an AgentSend inbox.

    Args:
        inbox_id: The AgentSend inbox ID to check for new messages.
    """
    resp = requests.get(
        f"{AGENTSEND_BASE}/inboxes/{inbox_id}/messages",
        headers={"Authorization": f"Bearer {AGENTSEND_API_KEY}"},
    )
    resp.raise_for_status()
    messages = resp.json().get("messages", [])
    if not messages:
        return "No new messages."
    lines = []
    for msg in messages[:5]:
        lines.append(
            f"From: {msg['from']}, Subject: {msg['subject']}, "
            f"Thread: {msg['threadId']}"
        )
    return "\n".join(lines)


@function_tool
def get_thread(thread_id: str) -> str:
    """Get all messages in an email thread.

    Args:
        thread_id: The thread ID to retrieve conversation history for.
    """
    resp = requests.get(
        f"{AGENTSEND_BASE}/threads/{thread_id}/messages",
        headers={"Authorization": f"Bearer {AGENTSEND_API_KEY}"},
    )
    resp.raise_for_status()
    data = resp.json()
    lines = [f"Thread: {data['subject']}"]
    for msg in data["messages"]:
        lines.append(f"[{msg['from']}]: {msg['text']}")
    return "\n".join(lines)

import asyncio
from agents import Agent, Runner
from email_tools import send_email, check_inbox, get_thread

email_agent = Agent(
    name="Email Assistant",
    instructions=(
        "You are a helpful assistant with email capabilities. "
        "Your inbox ID is inbox_abc123. You can send emails, "
        "check for new messages, and read thread history. "
        "Always use thread_id when replying to keep conversations grouped."
    ),
    tools=[send_email, check_inbox, get_thread],
)


async def main():
    result = await Runner.run(
        email_agent,
        "Check my inbox for new messages and summarize them.",
    )
    print(result.final_output)


asyncio.run(main())

from agents import Agent

email_agent = Agent(
    name="Email Agent",
    instructions="You handle all email communication. "
                  "Inbox ID: inbox_abc123.",
    tools=[send_email, check_inbox, get_thread],
)

main_agent = Agent(
    name="Support Manager",
    instructions=(
        "You manage customer support. When you need to "
        "send or check email, hand off to the Email Agent."
    ),
    handoffs=[email_agent],
)

from fastapi import FastAPI, Request
from agents import Runner

app = FastAPI()

@app.post("/webhook/email")
async def handle_email(request: Request):
    payload = await request.json()
    msg = payload["message"]

    result = await Runner.run(
        email_agent,
        f"New email from {msg['from']}. "
        f"Subject: {msg['subject']}. "
        f"Body: {msg['text']}. "
        f"Thread: {msg['threadId']}. "
        f"Reply appropriately.",
    )
    return {"status": "processed", "output": result.final_output}