Microsoft's AutoGen is one of the most popular frameworks for building multi-agent systems in Python. It handles the hard parts of agent orchestration — conversation management, tool execution, human-in-the-loop patterns — and lets you focus on defining what your agents actually do.
But AutoGen agents can't send or receive email out of the box. And if your agents interact with the real world — signing up for services, handling 2FA, sending reports, processing incoming messages — they need email. This tutorial shows you exactly how to add email capabilities to AutoGen using Lumbox, with complete, runnable code examples.
What We're Building
By the end of this tutorial, you'll have:
- A set of reusable Lumbox tool functions that any AutoGen agent can call
- A multi-agent workflow with two agents: a browser agent that navigates websites and a email agent that handles all email operations
- Isolated inboxes per agent to prevent cross-contamination
- A working example that signs up for a service, verifies the email, and confirms success
Prerequisites
You'll need Python 3.10+ and the following packages:
pip install autogen-agentchat autogen-ext lumbox playwright
playwright install chromium
You'll also need a Lumbox API key (sign up at lumbox.dev) and an OpenAI or Anthropic API key for the LLM.
Step 1: Define Lumbox Tool Functions
AutoGen agents use tools (functions) to interact with the world. We need to define tool functions that wrap the Lumbox SDK. These functions must be plain Python functions with type annotations — AutoGen uses the annotations to generate the tool schemas automatically.
import os
from typing import Optional
from lumbox import Lumbox
# Initialize the Lumbox client
lumbox_client = Lumbox(api_key=os.environ["LUMBOX_API_KEY"])
# Store for managing inboxes across function calls
inbox_store: dict[str, dict] = {}
def create_email_inbox(label: str) -> dict:
"""Create a new email inbox for an agent to use.
Args:
label: A human-readable label for this inbox (e.g., 'signup-flow-1')
Returns:
Dict with inbox_id and email address
"""
inbox = lumbox_client.inboxes.create()
inbox_store[label] = {
"id": inbox.id,
"email": inbox.email,
}
return {
"inbox_id": inbox.id,
"email": inbox.email,
"label": label,
}
def get_inbox_email(label: str) -> str:
"""Get the email address for a previously created inbox.
Args:
label: The label used when creating the inbox
Returns:
The email address string
"""
if label not in inbox_store:
return f"Error: No inbox found with label '{label}'"
return inbox_store[label]["email"]
def wait_for_otp_code(
label: str, timeout_seconds: int = 60
) -> dict:
"""Wait for an OTP verification code to arrive in an inbox.
Args:
label: The label of the inbox to monitor
timeout_seconds: How long to wait before giving up
Returns:
Dict with the OTP code, or an error message
"""
if label not in inbox_store:
return {"error": f"No inbox found with label '{label}'"}
inbox_id = inbox_store[label]["id"]
try:
otp = lumbox_client.messages.wait_for_otp(
inbox_id=inbox_id,
timeout=timeout_seconds * 1000, # Convert to ms
)
return {
"code": otp.code,
"from": otp.sender,
"subject": otp.subject,
}
except TimeoutError:
return {"error": "Timed out waiting for OTP"}
def list_inbox_emails(label: str) -> list[dict]:
"""List all emails in an inbox.
Args:
label: The label of the inbox
Returns:
List of email summaries
"""
if label not in inbox_store:
return [{"error": f"No inbox found with label '{label}'"}]
inbox_id = inbox_store[label]["id"]
emails = lumbox_client.inboxes.listEmails(inbox_id)
return [
{
"from": email.sender,
"subject": email.subject,
"received_at": str(email.received_at),
"snippet": email.text[:200] if email.text else "",
}
for email in emails
]
def send_email(
label: str,
to: str,
subject: str,
body_html: str,
) -> dict:
"""Send an email from an agent's inbox.
Args:
label: The label of the inbox to send from
to: Recipient email address
subject: Email subject line
body_html: HTML body of the email
Returns:
Confirmation dict
"""
if label not in inbox_store:
return {"error": f"No inbox found with label '{label}'"}
inbox_id = inbox_store[label]["id"]
lumbox_client.inboxes.send(inbox_id, {
"to": to,
"subject": subject,
"html": body_html,
})
return {
"status": "sent",
"from": inbox_store[label]["email"],
"to": to,
"subject": subject,
}
A few things to notice about these function definitions:
- Each function has a detailed docstring. AutoGen passes these to the LLM so the agent understands what each tool does and when to use it.
- We use a label-based store instead of passing raw inbox IDs around. This makes it easier for the LLM to reference inboxes by name ("use the signup inbox") rather than by opaque ID.
- All functions return dicts or simple types — AutoGen serializes these to pass between agents.
Step 2: Create the AutoGen Agents
Now we define two agents that work together. The email agent handles all email operations, and the coordinator agent orchestrates the overall workflow and decides when email actions are needed.
import autogen
# LLM configuration
llm_config = {
"config_list": [
{
"model": "claude-sonnet-4-20250514",
"api_key": os.environ["ANTHROPIC_API_KEY"],
"api_type": "anthropic",
}
],
"temperature": 0,
}
# The email agent has access to all Lumbox tools
email_agent = autogen.AssistantAgent(
name="email_agent",
system_message="""You are an email management agent. You handle all
email-related operations: creating inboxes, waiting for verification
codes, reading emails, and sending messages.
When asked to create an inbox, always use a descriptive label.
When waiting for an OTP, report the code clearly.
When listing emails, summarize the important details.
You have access to these tools:
- create_email_inbox: Create a new inbox
- get_inbox_email: Get an inbox's email address
- wait_for_otp_code: Wait for a verification code
- list_inbox_emails: List emails in an inbox
- send_email: Send an email from an inbox
Always confirm the result of each operation.""",
llm_config=llm_config,
)
# Register the tools with the email agent
email_agent.register_for_llm(
name="create_email_inbox",
description="Create a new email inbox for agent use",
)(create_email_inbox)
email_agent.register_for_llm(
name="get_inbox_email",
description="Get the email address for a labeled inbox",
)(get_inbox_email)
email_agent.register_for_llm(
name="wait_for_otp_code",
description="Wait for an OTP verification code in an inbox",
)(wait_for_otp_code)
email_agent.register_for_llm(
name="list_inbox_emails",
description="List all emails in a labeled inbox",
)(list_inbox_emails)
email_agent.register_for_llm(
name="send_email",
description="Send an email from a labeled inbox",
)(send_email)
# The coordinator orchestrates the workflow
coordinator = autogen.AssistantAgent(
name="coordinator",
system_message="""You are a workflow coordinator. You manage
multi-step processes that involve web browsing and email.
When a task requires email (signup, verification, sending messages),
delegate to the email_agent. When you need to coordinate between
steps, clearly state what needs to happen next.
Be concise and action-oriented.""",
llm_config=llm_config,
)
# A proxy agent that can execute the tool functions
user_proxy = autogen.UserProxyAgent(
name="executor",
human_input_mode="NEVER",
code_execution_config=False,
)
# Register tool execution with the proxy
user_proxy.register_for_execution(
name="create_email_inbox",
)(create_email_inbox)
user_proxy.register_for_execution(
name="get_inbox_email",
)(get_inbox_email)
user_proxy.register_for_execution(
name="wait_for_otp_code",
)(wait_for_otp_code)
user_proxy.register_for_execution(
name="list_inbox_emails",
)(list_inbox_emails)
user_proxy.register_for_execution(
name="send_email",
)(send_email)
Step 3: Build the Multi-Agent Workflow
Let's build a real workflow: signing up for a web service that requires email verification. We'll use AutoGen's GroupChat to coordinate the agents.
def run_signup_workflow(service_url: str, password: str):
"""Run a multi-agent signup workflow with email verification."""
group_chat = autogen.GroupChat(
agents=[coordinator, email_agent, user_proxy],
messages=[],
max_round=20,
speaker_selection_method="auto",
)
manager = autogen.GroupChatManager(
groupchat=group_chat,
llm_config=llm_config,
)
# Kick off the workflow
task = f"""Complete this multi-step signup workflow:
1. First, ask email_agent to create a new inbox labeled 'signup'.
2. Once you have the email address, report it back. (In production,
a browser agent would use this to fill the signup form at
{service_url})
3. After the form would be submitted, ask email_agent to wait for
an OTP code in the 'signup' inbox.
4. Once you have the OTP, report it. (The browser agent would enter
this code.)
5. Finally, ask email_agent to send a confirmation email from the
'signup' inbox to admin@example.com with the subject
'Signup Complete' confirming the process finished.
Execute each step and report results clearly."""
user_proxy.initiate_chat(manager, message=task)
Step 4: Isolated Inboxes Per Agent
In multi-agent systems, isolation is critical. If Agent A and Agent B both need email, they should each have their own inbox. This prevents agents from reading each other's messages and eliminates race conditions when multiple agents wait for OTPs simultaneously.
Here's a pattern for managing per-agent inboxes:
class AgentInboxManager:
"""Manages isolated email inboxes for each agent in a multi-agent system."""
def __init__(self, api_key: str):
self.client = Lumbox(api_key=api_key)
self.agent_inboxes: dict[str, list[dict]] = {}
def create_inbox_for_agent(
self, agent_name: str, purpose: str
) -> dict:
"""Create a dedicated inbox for a specific agent.
Args:
agent_name: The name of the agent
purpose: What this inbox will be used for
Returns:
Inbox details including id and email
"""
inbox = self.client.inboxes.create()
inbox_info = {
"id": inbox.id,
"email": inbox.email,
"purpose": purpose,
"agent": agent_name,
}
if agent_name not in self.agent_inboxes:
self.agent_inboxes[agent_name] = []
self.agent_inboxes[agent_name].append(inbox_info)
return inbox_info
def get_agent_inboxes(self, agent_name: str) -> list[dict]:
"""Get all inboxes belonging to an agent."""
return self.agent_inboxes.get(agent_name, [])
def wait_for_otp(
self, agent_name: str, purpose: str, timeout: int = 60_000
) -> Optional[str]:
"""Wait for OTP in a specific agent's inbox.
Args:
agent_name: Which agent's inbox to check
purpose: Which inbox (by purpose) to monitor
timeout: Timeout in milliseconds
Returns:
The OTP code, or None
"""
inboxes = self.agent_inboxes.get(agent_name, [])
target = next(
(i for i in inboxes if i["purpose"] == purpose),
None,
)
if not target:
return None
otp = self.client.messages.wait_for_otp(
inbox_id=target["id"],
timeout=timeout,
)
return otp.code if otp else None
# Usage in a multi-agent system
inbox_mgr = AgentInboxManager(os.environ["LUMBOX_API_KEY"])
# Each agent gets its own inbox
browser_inbox = inbox_mgr.create_inbox_for_agent(
"browser_agent", "web-signups"
)
research_inbox = inbox_mgr.create_inbox_for_agent(
"research_agent", "newsletter-subscriptions"
)
# Agents can only access their own inboxes
# browser_agent uses browser_inbox.email for signups
# research_agent uses research_inbox.email for newsletters
# No cross-contamination possible
Step 5: Complete Working Example
Here's a complete, end-to-end example that ties everything together. This creates a two-agent system where one agent handles web signup and another handles email verification, demonstrating the full pattern:
import os
import autogen
from lumbox import Lumbox
# ---- Configuration ----
lumbox_client = Lumbox(api_key=os.environ["LUMBOX_API_KEY"])
inbox_store: dict[str, dict] = {}
llm_config = {
"config_list": [
{
"model": "claude-sonnet-4-20250514",
"api_key": os.environ["ANTHROPIC_API_KEY"],
"api_type": "anthropic",
}
],
"temperature": 0,
}
# ---- Tool Functions ----
def create_inbox(label: str) -> dict:
"""Create a new email inbox with the given label."""
inbox = lumbox_client.inboxes.create()
inbox_store[label] = {"id": inbox.id, "email": inbox.email}
return {"label": label, "email": inbox.email, "inbox_id": inbox.id}
def wait_for_code(label: str, timeout_seconds: int = 60) -> dict:
"""Wait for a verification code in the labeled inbox."""
if label not in inbox_store:
return {"error": f"No inbox '{label}'"}
try:
otp = lumbox_client.messages.wait_for_otp(
inbox_id=inbox_store[label]["id"],
timeout=timeout_seconds * 1000,
)
return {"code": otp.code}
except Exception as e:
return {"error": str(e)}
def read_emails(label: str) -> list[dict]:
"""Read all emails in the labeled inbox."""
if label not in inbox_store:
return [{"error": f"No inbox '{label}'"}]
emails = lumbox_client.inboxes.listEmails(inbox_store[label]["id"])
return [
{"from": e.sender, "subject": e.subject, "text": e.text[:300]}
for e in emails
]
def send_from_inbox(
label: str, to: str, subject: str, html: str
) -> dict:
"""Send an email from the labeled inbox."""
if label not in inbox_store:
return {"error": f"No inbox '{label}'"}
lumbox_client.inboxes.send(
inbox_store[label]["id"],
{"to": to, "subject": subject, "html": html},
)
return {"status": "sent", "to": to}
# ---- Agent Setup ----
email_agent = autogen.AssistantAgent(
name="EmailAgent",
system_message=(
"You manage email operations. Use your tools to create "
"inboxes, wait for verification codes, read emails, and "
"send messages. Always report results clearly."
),
llm_config=llm_config,
)
for fn in [create_inbox, wait_for_code, read_emails, send_from_inbox]:
email_agent.register_for_llm(
name=fn.__name__,
description=fn.__doc__,
)(fn)
executor = autogen.UserProxyAgent(
name="Executor",
human_input_mode="NEVER",
code_execution_config=False,
)
for fn in [create_inbox, wait_for_code, read_emails, send_from_inbox]:
executor.register_for_execution(name=fn.__name__)(fn)
coordinator = autogen.AssistantAgent(
name="Coordinator",
system_message=(
"You coordinate multi-step workflows involving web "
"actions and email. Delegate email tasks to EmailAgent. "
"Be concise and track progress through each step."
),
llm_config=llm_config,
)
# ---- Run the Workflow ----
def main():
group_chat = autogen.GroupChat(
agents=[coordinator, email_agent, executor],
messages=[],
max_round=15,
)
manager = autogen.GroupChatManager(
groupchat=group_chat,
llm_config=llm_config,
)
task = """
Execute this workflow:
1. Create an email inbox labeled 'demo-signup'
2. Report the email address (simulating form fill)
3. Wait for a verification code in that inbox
4. Report the code (simulating code entry)
5. Send a summary email from 'demo-signup' to
team@example.com with subject 'Agent Signup Complete'
"""
executor.initiate_chat(manager, message=task)
if __name__ == "__main__":
main()
Error Handling and Resilience
Production multi-agent systems need robust error handling. Here's how to add retries and fallback logic to the email tools:
import time
from functools import wraps
def with_retry(max_attempts: int = 3, delay: float = 2.0):
"""Decorator that retries a function on failure."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_error = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
last_error = e
if attempt < max_attempts - 1:
time.sleep(delay * (attempt + 1))
return {"error": f"Failed after {max_attempts} attempts: {last_error}"}
return wrapper
return decorator
@with_retry(max_attempts=3)
def wait_for_code_resilient(label: str, timeout_seconds: int = 60) -> dict:
"""Wait for a verification code with automatic retry."""
if label not in inbox_store:
raise ValueError(f"No inbox '{label}'")
otp = lumbox_client.messages.wait_for_otp(
inbox_id=inbox_store[label]["id"],
timeout=timeout_seconds * 1000,
)
if not otp or not otp.code:
raise Exception("No OTP code found in received email")
return {"code": otp.code}
You should also handle the case where an OTP expires before the agent can use it. If the browser agent is slow to enter the code, the 2FA might time out. Build in a retry loop at the workflow level:
async def verified_login_flow(page, inbox_label: str, max_retries: int = 3):
"""Attempt login with email verification, retrying if OTP expires."""
for attempt in range(max_retries):
# Submit login form (triggers OTP send)
await page.click('button[type="submit"]')
# Wait for OTP
result = wait_for_code(inbox_label, timeout_seconds=90)
if "error" in result:
print(f"Attempt {attempt + 1}: OTP retrieval failed, retrying...")
# Some services have a "resend code" button
resend_btn = page.locator('button:has-text("Resend")')
if await resend_btn.is_visible():
await resend_btn.click()
continue
# Enter OTP
await page.fill('[data-testid="otp-input"]', result["code"])
await page.click('button[type="submit"]')
# Check if login succeeded
try:
await page.wait_for_url("**/dashboard**", timeout=10_000)
return True
except Exception:
print(f"Attempt {attempt + 1}: OTP may have expired, retrying...")
continue
return False
Tips for Production Deployment
A few lessons from running this pattern in production:
1. Clean Up Inboxes
Don't leave inboxes around forever. Create them at the start of a workflow and clean them up when the workflow completes (or fails). Orphaned inboxes are a waste of resources and a potential security concern.
2. Log Everything
Email operations are asynchronous and time-sensitive. Log when you create an inbox, when you start waiting for an OTP, when the OTP arrives, and when you use it. This is invaluable for debugging when an agent flow fails at the verification step.
3. Set Reasonable Timeouts
Sixty seconds is a good default for OTP waiting. Some services are slower — particularly enterprise tools that route through internal mail servers. If you're consistently timing out, increase to 90 or 120 seconds. But don't set it to 5 minutes — if the email hasn't arrived in 2 minutes, something is wrong and you should retry the whole flow.
4. Use Labels Consistently
The label system in our implementation isn't just for convenience — it prevents a class of bugs where agents reference the wrong inbox by ID. Labels like `"signup-acme-corp"` or `"2fa-salesforce"` are self-documenting and make logs much easier to read.
5. Test with Real Services
Don't just test with mocked email. Real email services send wildly different HTML templates, some with the OTP in a `<span>`, others in a `<td>`, others as plain text. Lumbox's OTP extraction handles this variety, but you should verify with the specific services your agent targets.
What's Next
This tutorial covers the core pattern: AutoGen agents with email capabilities via Lumbox. From here, you can extend in several directions:
- Add browser automation agents that use Playwright tools alongside the email tools, creating a fully autonomous web workflow
- Build specialized email processing agents that read incoming emails, extract structured data, and trigger downstream workflows
- Implement approval flows where an agent sends an email to a human, waits for a reply, and continues based on the response
- Expose the email tools as an MCP server so any MCP-compatible agent framework can use them, not just AutoGen
The combination of AutoGen's multi-agent orchestration and Lumbox's email primitives gives you the building blocks for agents that actually interact with the real world — not just chat interfaces, but systems that sign up for services, handle verification, send and receive messages, and complete end-to-end workflows autonomously.