send and receive email from AutoGen agents

pip install autogen-agentchat autogen-ext[openai] httpx

import httpx

BASE_URL = "https://api.lobstermail.ai"

async def signup_for_email() -> dict:
    """Create a new LobsterMail account. Returns an auth token
    and account ID that the agent uses for all subsequent calls."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(f"{BASE_URL}/v1/signup")
        resp.raise_for_status()
        return resp.json()
        # {"token": "lm_sk_live_...", "accountId": "..."}

async def create_inbox(token: str, handle: str) -> dict:
    """Provision a new email inbox with the given handle.
    Returns the full email address and inbox ID."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{BASE_URL}/v1/inboxes",
            headers={"Authorization": f"Bearer {token}"},
            json={"name": handle},
        )
        resp.raise_for_status()
        return resp.json()
        # {"inboxId": "...", "address": "handle@lobstermail.ai"}

async def send_email(
    token: str, from_addr: str, to_addr: str, subject: str, body: str
) -> dict:
    """Send an email from a LobsterMail inbox."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{BASE_URL}/v1/emails/send",
            headers={"Authorization": f"Bearer {token}"},
            json={
                "from": from_addr,
                "to": to_addr,
                "subject": subject,
                "body": body,
            },
        )
        resp.raise_for_status()
        return resp.json()

async def check_inbox(token: str, inbox_id: str) -> dict:
    """Fetch recent emails for a given inbox."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{BASE_URL}/v1/inboxes/{inbox_id}/emails",
            headers={"Authorization": f"Bearer {token}"},
        )
        resp.raise_for_status()
        return resp.json()

from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.tools import FunctionTool

# Wrap each function as an AutoGen tool
signup_tool = FunctionTool(signup_for_email, description="Create a new LobsterMail account to get an auth token")
inbox_tool = FunctionTool(create_inbox, description="Provision a new email inbox with a chosen handle")
send_tool = FunctionTool(send_email, description="Send an email from a LobsterMail inbox")
check_tool = FunctionTool(check_inbox, description="Check a LobsterMail inbox for new emails")

model_client = OpenAIChatCompletionClient(model="gpt-4o")

agent = AssistantAgent(
    name="email_agent",
    model_client=model_client,
    tools=[signup_tool, inbox_tool, send_tool, check_tool],
    system_message=(
        "You are an email assistant. Use the LobsterMail tools to manage email. "
        "First sign up to get a token, then create an inbox, then send or check "
        "emails as requested. Always pass the token from signup to subsequent calls."
    ),
)

import asyncio
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken

async def main():
    response = await agent.on_messages(
        [TextMessage(
            content="Sign up for email, create an inbox called 'outreach-bot', "
                    "then send a message to hello@example.com introducing yourself.",
            source="user",
        )],
        cancellation_token=CancellationToken(),
    )
    print(response.chat_message.content)

asyncio.run(main())

research_agent = AssistantAgent(
    name="researcher",
    model_client=model_client,
    tools=[signup_tool, inbox_tool, send_tool, check_tool],
    system_message="You research topics and email your findings to the writer agent.",
)

writer_agent = AssistantAgent(
    name="writer",
    model_client=model_client,
    tools=[signup_tool, inbox_tool, send_tool, check_tool],
    system_message="You receive research via email and draft polished content.",
)

import asyncio

async def poll_and_respond(agent, token, inbox_id):
    seen = set()
    while True:
        emails = await check_inbox(token, inbox_id)
        for email in emails.get("emails", []):
            if email["id"] not in seen:
                seen.add(email["id"])
                # Feed the email into the agent as a new message
                response = await agent.on_messages(
                    [TextMessage(
                        content=f"New email from {email['from']}: "
                                f"Subject: {email['subject']}\n\n{email['body']}",
                        source="email_webhook",
                    )],
                    cancellation_token=CancellationToken(),
                )
                print(response.chat_message.content)
        await asyncio.sleep(30)

async def safe_check_inbox(token: str, inbox_id: str) -> dict:
    """Fetch emails, filtering out injection risks."""
    result = await check_inbox(token, inbox_id)
    safe_emails = [
        e for e in result.get("emails", [])
        if not e.get("isInjectionRisk", False)
    ]
    return {"emails": safe_emails}