AutoGen 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": "AutoGen Outreach",
    "emailAddress": "agent@yourdomain.com"
  }'

import os
import requests
from typing import Annotated


def send_email(
    to: Annotated[str, "Recipient email address"],
    subject: Annotated[str, "Email subject line"],
    body: Annotated[str, "Email body in plain text or HTML"],
    dedupe_key: Annotated[
        str,
        "Unique key to prevent duplicate sends. Use a descriptive value like 'welcome-user@example.com'.",
    ],
) -> str:
    """
    Send an email through the Molted policy engine. The email is checked against
    20+ policy rules before delivery. Returns the requestId on success or the
    block reason if rejected. Do not retry blocked sends.
    """
    response = requests.post(
        "https://api.molted.email/v1/agent/send/request",
        headers={
            "Authorization": f"Bearer {os.environ['MOLTED_API_KEY']}",
            "Content-Type": "application/json",
        },
        json={
            "tenantId": os.environ["MOLTED_TENANT_ID"],
            "recipientEmail": to,
            "templateId": "_default",
            "dedupeKey": dedupe_key,
            "agentId": "autogen-agent",
            "payload": {
                "subject": subject,
                "html": body,
                "text": body,
            },
        },
        timeout=10,
    )
    data = response.json()

    if data.get("status") == "blocked":
        return (
            f"BLOCKED: {data.get('blockReason')}. "
            f"Decision trace ID: {data.get('requestId')}. "
            "Do not retry this send."
        )

    return f"Queued: requestId={data.get('requestId')}, status={data.get('status')}"


def check_email_status(
    request_id: Annotated[str, "The requestId returned when the email was queued"],
) -> str:
    """Check the delivery status of a previously queued email."""
    response = requests.get(
        f"https://api.molted.email/v1/agent/send/{request_id}/status",
        headers={"Authorization": f"Bearer {os.environ['MOLTED_API_KEY']}"},
        timeout=10,
    )
    data = response.json()

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

import os
from autogen import AssistantAgent, UserProxyAgent
from email_tools import send_email, check_email_status

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": os.environ["OPENAI_API_KEY"]}],
    "temperature": 0,
}

assistant = AssistantAgent(
    name="EmailAgent",
    system_message=(
        "You are an email outreach agent. You send policy-checked emails on behalf of "
        "the user. When sending emails:\n"
        "- Always provide a descriptive dedupeKey (e.g. 'welcome-alice@example.com') "
        "  to prevent accidental duplicate sends.\n"
        "- If a send returns BLOCKED, report the reason to the user and do not retry.\n"
        "- After sending, confirm the requestId to the user.\n"
        "- Do not attempt to send to suppressed or opted-out contacts."
    ),
    llm_config=llm_config,
)

user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=5,
    code_execution_config=False,
)

# Register the send and status tools
assistant.register_for_llm(name="send_email", description=send_email.__doc__)(send_email)
assistant.register_for_llm(name="check_email_status", description=check_email_status.__doc__)(check_email_status)

user_proxy.register_for_execution(name="send_email")(send_email)
user_proxy.register_for_execution(name="check_email_status")(check_email_status)

# Start a conversation
user_proxy.initiate_chat(
    assistant,
    message=(
        "Send a trial welcome email to alice@example.com. "
        "Subject: 'Welcome to your trial'. "
        "Body: 'Your 14-day trial starts today. Reply if you have any questions.'"
    ),
)

import os
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
from email_tools import send_email, check_email_status

llm_config = {
    "config_list": [{"model": "gpt-4o", "api_key": os.environ["OPENAI_API_KEY"]}],
}

# Outreach agent - can send email
outreach_agent = AssistantAgent(
    name="OutreachAgent",
    system_message=(
        "You send policy-compliant outreach emails. "
        "Always use a descriptive dedupeKey. "
        "Report BLOCKED sends to the group without retrying."
    ),
    llm_config=llm_config,
)

# Research agent - no email access
research_agent = AssistantAgent(
    name="ResearchAgent",
    system_message=(
        "You research contacts and draft email content. "
        "You do not send emails - pass your drafts to OutreachAgent."
    ),
    llm_config=llm_config,
)

user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="TERMINATE",
    code_execution_config=False,
)

# Only register email tools on the outreach agent and executor
outreach_agent.register_for_llm(name="send_email", description=send_email.__doc__)(send_email)
outreach_agent.register_for_llm(name="check_email_status", description=check_email_status.__doc__)(check_email_status)
user_proxy.register_for_execution(name="send_email")(send_email)
user_proxy.register_for_execution(name="check_email_status")(check_email_status)

group_chat = GroupChat(
    agents=[user_proxy, research_agent, outreach_agent],
    messages=[],
    max_round=10,
)

manager = GroupChatManager(groupchat=group_chat, llm_config=llm_config)

user_proxy.initiate_chat(
    manager,
    message="Research and send a personalized trial invitation to bob@example.com who leads engineering at Acme Corp.",
)

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_core.tools import FunctionTool
from autogen_ext.models.openai import OpenAIChatCompletionClient
from email_tools import send_email, check_email_status

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

send_tool = FunctionTool(send_email, description=send_email.__doc__ or "Send a policy-checked email")
status_tool = FunctionTool(check_email_status, description=check_email_status.__doc__ or "Check email delivery status")

email_agent = AssistantAgent(
    name="EmailAgent",
    model_client=model_client,
    tools=[send_tool, status_tool],
    system_message=(
        "You send policy-checked emails. Always include a descriptive dedupeKey. "
        "Report BLOCKED sends without retrying. "
        "Say 'TERMINATE' when the task is complete."
    ),
)

termination = TextMentionTermination("TERMINATE")
team = RoundRobinGroupChat([email_agent], termination_condition=termination)

import asyncio

async def main():
    result = await team.run(
        task="Send a welcome email to carol@example.com. Subject: 'Welcome!' Body: 'Thanks for signing up.'"
    )
    print(result)

asyncio.run(main())

import requests
import os

agents_to_register = [
    {"name": "outreach-agent"},
    {"name": "research-agent"},
]

for agent in agents_to_register:
    response = requests.post(
        "https://api.molted.email/v1/agent/register",
        headers={
            "Authorization": f"Bearer {os.environ['MOLTED_API_KEY']}",
            "Content-Type": "application/json",
        },
        json={"tenantId": os.environ["MOLTED_TENANT_ID"], **agent},
    )
    data = response.json()
    print(f"Registered {agent['name']}: agentId={data['id']}")