GPT-4o 에이전트의 이메일 송수신: AgentMail 웹훅 기반 회신 처리 패턴

from openai import OpenAI
import requests, os, json

client   = OpenAI()
AM_KEY   = os.environ["AGENTMAIL_KEY"]
INBOX_ID = os.environ["INBOX_ID"]   # created once via POST /inboxes
H        = {"Authorization": f"Bearer {AM_KEY}"}

tools = [
    {
        "type": "function",
        "function": {
            "name": "send_email",
            "description": "Send an email or reply in an existing thread.",
            "parameters": {
                "type": "object",
                "properties": {
                    "to":        {"type": "string"},
                    "subject":   {"type": "string"},
                    "body":      {"type": "string"},
                    "thread_id": {"type": "string", "description": "Pass to reply in existing thread"}
                },
                "required": ["to", "subject", "body"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "read_thread",
            "description": "Fetch the full thread for context before replying.",
            "parameters": {
                "type": "object",
                "properties": {
                    "thread_id": {"type": "string"}
                },
                "required": ["thread_id"]
            }
        }
    }
]

def send_email(to, subject, body, thread_id=None):
    payload = {"to": [to], "subject": subject, "text": body}
    if thread_id:
        payload["thread_id"] = thread_id
    return requests.post(
        f"https://api.agentmail.to/v0/inboxes/{INBOX_ID}/emails",
        headers=H, json=payload
    ).json()

def read_thread(thread_id):
    msgs = requests.get(
        f"https://api.agentmail.to/v0/threads/{thread_id}",
        headers=H
    ).json().get("messages", [])
    return "
---
".join(f"From: {m['from']}
{m['text']}" for m in msgs)

# Webhook fires when a reply arrives — agent wakes up in <5 seconds
def handle_reply(event: dict):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": (
                f"Reply from {event['from']}. "
                f"Thread ID: {event['thread_id']}. "
                f"Their message: {event['text']}. "
                "Read the thread for context and respond."
            )
        }],
        tools=tools
    )

    for tool_call in response.choices[0].message.tool_calls or []:
        args = json.loads(tool_call.function.arguments)
        if tool_call.function.name == "send_email":
            send_email(**args)
        elif tool_call.function.name == "read_thread":
            context = read_thread(**args["thread_id"])
            # Feed back into next completion for full context