Email APIs for AI Agents with FastAPI

import httpx
from fastapi import FastAPI, Depends
from pydantic import BaseModel
from typing import Optional
import os

app = FastAPI(title="Email Agent API")

MULTIMAIL_API = "https://api.multimail.dev/v1"

async def get_multimail_client():
    async with httpx.AsyncClient(
        base_url=MULTIMAIL_API,
        headers={"Authorization": f"Bearer {os.environ[&"cm">#039;MULTIMAIL_API_KEY']}"}
    ) as client:
        yield client


class SendEmailRequest(BaseModel):
    mailbox_id: str
    to: str
    subject: str
    body: str


class EmailMessage(BaseModel):
    id: str
    from_address: str
    subject: str
    body: str
    received_at: str
    thread_id: Optional[str] = None

@app.post("/send")
async def send_email(
    request: SendEmailRequest,
    client: httpx.AsyncClient = Depends(get_multimail_client)
):
    """Send an email. In gated_send mode, queued for human approval."""
    resp = await client.post("/send", json=request.model_dump())
    resp.raise_for_status()
    return resp.json()


@app.get("/inbox/{mailbox_id}")
async def check_inbox(
    mailbox_id: str,
    limit: int = 10,
    client: httpx.AsyncClient = Depends(get_multimail_client)
):
    """Check a mailbox inbox for recent messages."""
    resp = await client.get(
        f"/mailboxes/{mailbox_id}/inbox",
        params={"limit": limit}
    )
    resp.raise_for_status()
    return resp.json()


@app.post("/reply")
async def reply_email(
    message_id: str,
    body: str,
    client: httpx.AsyncClient = Depends(get_multimail_client)
):
    """Reply to an email within its thread."""
    resp = await client.post("/reply", json={
        "message_id": message_id,
        "body": body
    })
    resp.raise_for_status()
    return resp.json()

from fastapi import BackgroundTasks
from openai import AsyncOpenAI

openai_client = AsyncOpenAI()

async def process_incoming_email(
    message_id: str,
    from_addr: str,
    subject: str,
    body: str,
):
    """Background task: generate an AI reply and queue it via MultiMail."""
    completion = await openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Draft a brief, professional reply."},
            {"role": "user", "content": f"From: {from_addr}\nSubject: {subject}\n\n{body}"},
        ]
    )
    reply_body = completion.choices[0].message.content

    async with httpx.AsyncClient(
        base_url=MULTIMAIL_API,
        headers={"Authorization": f"Bearer {os.environ[&"cm">#039;MULTIMAIL_API_KEY']}"}
    ) as client:
        await client.post("/reply", json={
            "message_id": message_id,
            "body": reply_body
        })


@app.post("/webhook/email-received")
async def handle_email_webhook(
    payload: dict,
    background_tasks: BackgroundTasks
):
    """Webhook endpoint for MultiMail email.received events."""
    if payload.get("event") == "email.received":
        background_tasks.add_task(
            process_incoming_email,
            payload["message_id"],
            payload["from"],
            payload["subject"],
            payload["body"],
        )
    return {"status": "accepted"}

@app.get("/threads/{thread_id}")
async def get_thread(
    thread_id: str,
    client: httpx.AsyncClient = Depends(get_multimail_client)
):
    """Get all messages in an email thread."""
    resp = await client.get(f"/threads/{thread_id}")
    resp.raise_for_status()
    return resp.json()


@app.get("/mailboxes")
async def list_mailboxes(
    client: httpx.AsyncClient = Depends(get_multimail_client)
):
    """List all available mailboxes."""
    resp = await client.get("/mailboxes")
    resp.raise_for_status()
    return resp.json()