Email infrastructure for Google ADK agents

Give your ADK agents a supervised email channel — send, read, and approve messages through MultiMail's REST API with graduated oversight that fits ADK's tool-calling model.

Google ADK structures agents as tool-calling systems: each capability is a typed Python function the agent can invoke during a session. Email is a natural fit — your agent can check an inbox, read a thread, draft a reply, or send a new message as first-class tool calls, not raw SMTP calls embedded in the prompt.

MultiMail is designed for exactly this pattern. Its REST API exposes each email operation as a discrete endpoint — send_email, check_inbox, read_email, reply_email, decide_email — that maps cleanly to an ADK tool definition. Each call carries your API key, and each mailbox has an oversight mode that governs how much the agent can do without human approval.

The combination is particularly useful for long-running ADK workflows that end in outbound communication. Rather than giving the agent direct SMTP access, you route email through MultiMail and choose an oversight mode: gated_send holds every outbound message for human approval before delivery, while monitored lets the agent act freely and notifies a human on every send.

Why MultiMail

Built for Google ADK (Agent Development Kit) developers

Tool-shaped API

Every MultiMail operation maps to one ADK tool definition. send_email, check_inbox, read_email, and decide_email are individual REST endpoints, not a general-purpose SMTP wrapper. ADK's type-annotated function tools translate directly — no adapter layer needed.

Oversight modes that fit ADK's async session model

ADK sessions can span many steps before producing outbound communication. MultiMail's gated_send mode holds the final send for approval without blocking the session — the agent keeps working while a human reviews the draft in a separate flow.

Per-agent email identity

Each ADK agent can be assigned its own MultiMail mailbox ([email protected], [email protected]). Recipients see a distinct sender address per agent, and you can apply different oversight modes to each mailbox independently.

Approval webhooks for human-in-the-loop

When a message is held for approval in gated_send mode, MultiMail fires a webhook. Wire this to an ADK evaluation step or a human callback without polling list_pending on a timer.

Test key separation for evaluation runs

ADK evaluation workflows use MultiMail test keys (mm_test_...) that record sends without delivering them. Switch to mm_live_... for production. Tool function signatures are identical in both modes — just swap the environment variable.

Code examples

Get started in minutes

Define MultiMail tools for an ADK agent
python
import os
import requests
from google.adk.agents import Agent

MULTIMAIL_API = "https://api.multimail.dev"
HEADERS = {
    "Authorization": f"Bearer {os.environ[&"cm">#039;MULTIMAIL_API_KEY']}",
    "Content-Type": "application/json",
}


def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email from the agent&"cm">#039;s mailbox. Returns message_id and delivery status."""
    resp = requests.post(
        f"{MULTIMAIL_API}/send_email",
        headers=HEADERS,
        json={"to": to, "subject": subject, "body": body},
    )
    resp.raise_for_status()
    return resp.json()


def check_inbox(mailbox: str, unread_only: bool = True) -> dict:
    """List recent emails in a mailbox. Returns a list of message summaries with IDs."""
    resp = requests.get(
        f"{MULTIMAIL_API}/check_inbox",
        headers=HEADERS,
        params={"mailbox": mailbox, "unread_only": unread_only},
    )
    resp.raise_for_status()
    return resp.json()


def read_email(message_id: str) -> dict:
    """Fetch the full body and metadata of a single email by message_id."""
    resp = requests.get(
        f"{MULTIMAIL_API}/read_email",
        headers=HEADERS,
        params={"message_id": message_id},
    )
    resp.raise_for_status()
    return resp.json()


def reply_email(message_id: str, body: str) -> dict:
    """Reply to an existing email thread. Preserves thread context automatically."""
    resp = requests.post(
        f"{MULTIMAIL_API}/reply_email",
        headers=HEADERS,
        json={"message_id": message_id, "body": body},
    )
    resp.raise_for_status()
    return resp.json()


agent = Agent(
    model="gemini-2.0-flash",
    name="email_agent",
    instruction="You are an email assistant. Use the provided tools to read and respond to emails.",
    tools=[send_email, check_inbox, read_email, reply_email],
)

Wrap MultiMail REST endpoints as typed Python functions. ADK reads type annotations and docstrings to build the tool schema automatically — no decorator or registration call required.

Gated send — hold outbound messages for human approval
python
import os
import requests
from google.adk.agents import Agent

MULTIMAIL_API = "https://api.multimail.dev"
HEADERS = {
    "Authorization": f"Bearer {os.environ[&"cm">#039;MULTIMAIL_API_KEY']}",
    "Content-Type": "application/json",
}


def send_email(to: str, subject: str, body: str) -> dict:
    """Queue an email for human approval before delivery. Returns pending message_id."""
    resp = requests.post(
        f"{MULTIMAIL_API}/send_email",
        headers=HEADERS,
        json={"to": to, "subject": subject, "body": body},
    )
    resp.raise_for_status()
    "cm"># With gated_send oversight, result["status"] == "pending_approval"
    return resp.json()


def list_pending() -> dict:
    """List all outbound messages currently waiting for human approval."""
    resp = requests.get(f"{MULTIMAIL_API}/list_pending", headers=HEADERS)
    resp.raise_for_status()
    return resp.json()


agent = Agent(
    model="gemini-2.0-flash",
    name="gated_email_agent",
    instruction=(
        "Draft and queue outbound emails using send_email. "
        "Always confirm the recipient and subject before sending. "
        "Inform the user that messages require human approval before delivery."
    ),
    tools=[send_email, list_pending],
)


"cm"># Human-side approval — called from your webhook handler or approval UI
def decide_email(message_id: str, decision: str, reason: str = "") -> dict:
    """Approve or reject a pending message. decision must be &"cm">#039;approve' or 'reject'."""
    resp = requests.post(
        f"{MULTIMAIL_API}/decide_email",
        headers=HEADERS,
        json={"message_id": message_id, "decision": decision, "reason": reason},
    )
    resp.raise_for_status()
    return resp.json()

With gated_send oversight on the mailbox, send_email returns pending_approval immediately. The ADK session continues without blocking. A human approves or rejects via decide_email before delivery occurs.

Multi-step session: read, tag, and reply
python
import asyncio
import os
import requests
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

MULTIMAIL_API = "https://api.multimail.dev"
HEADERS = {
    "Authorization": f"Bearer {os.environ[&"cm">#039;MULTIMAIL_API_KEY']}",
    "Content-Type": "application/json",
}
MAILBOX = "[email protected]"


def check_inbox(unread_only: bool = True) -> dict:
    """List recent emails in the support mailbox. Returns message summaries with IDs."""
    resp = requests.get(
        f"{MULTIMAIL_API}/check_inbox",
        headers=HEADERS,
        params={"mailbox": MAILBOX, "unread_only": unread_only},
    )
    resp.raise_for_status()
    return resp.json()


def read_email(message_id: str) -> dict:
    """Fetch full email content by message_id."""
    resp = requests.get(
        f"{MULTIMAIL_API}/read_email",
        headers=HEADERS,
        params={"message_id": message_id},
    )
    resp.raise_for_status()
    return resp.json()


def tag_email(message_id: str, tags: list[str]) -> dict:
    """Apply classification tags to an email (e.g. billing, technical, general)."""
    resp = requests.post(
        f"{MULTIMAIL_API}/tag_email",
        headers=HEADERS,
        json={"message_id": message_id, "tags": tags},
    )
    resp.raise_for_status()
    return resp.json()


def reply_email(message_id: str, body: str) -> dict:
    """Reply to a message thread. Preserves thread context."""
    resp = requests.post(
        f"{MULTIMAIL_API}/reply_email",
        headers=HEADERS,
        json={"message_id": message_id, "body": body},
    )
    resp.raise_for_status()
    return resp.json()


agent = Agent(
    model="gemini-2.0-flash",
    name="support_triage_agent",
    instruction=(
        "Check the support inbox, read each unread email, tag it by category "
        "(billing, technical, or general), then draft and send a helpful reply "
        "using reply_email."
    ),
    tools=[check_inbox, read_email, tag_email, reply_email],
)


async def run_triage():
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="support_triage", user_id="system"
    )
    runner = Runner(
        agent=agent,
        app_name="support_triage",
        session_service=session_service,
    )
    async for event in runner.run_async(
        user_id="system",
        session_id=session.id,
        new_message=types.Content(
            role="user",
            parts=[types.Part(text="Process all unread support emails.")],
        ),
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)


asyncio.run(run_triage())

An ADK session that checks a support inbox, reads unread messages, tags each one by category, and drafts replies — using MultiMail tools across the full lifecycle.

Evaluation workflow with test API key
python
import os
import requests
from google.adk.agents import Agent
from google.adk.evaluation import AgentEvaluator

"cm"># mm_test_... keys record sends without delivering — safe for eval
API_KEY = os.environ.get("MULTIMAIL_API_KEY", "mm_test_your_test_key_here")
MULTIMAIL_API = "https://api.multimail.dev"
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}


def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email. In test mode, records the send without delivering it."""
    resp = requests.post(
        f"{MULTIMAIL_API}/send_email",
        headers=HEADERS,
        json={"to": to, "subject": subject, "body": body},
    )
    resp.raise_for_status()
    return resp.json()


def check_inbox(mailbox: str, unread_only: bool = True) -> dict:
    """List recent emails in a mailbox."""
    resp = requests.get(
        f"{MULTIMAIL_API}/check_inbox",
        headers=HEADERS,
        params={"mailbox": mailbox, "unread_only": unread_only},
    )
    resp.raise_for_status()
    return resp.json()


agent = Agent(
    model="gemini-2.0-flash",
    name="email_eval_agent",
    instruction="You are an email assistant. Use send_email and check_inbox to manage communications.",
    tools=[send_email, check_inbox],
)

"cm"># No real emails sent during evaluation
evaluator = AgentEvaluator()
evaluator.evaluate(
    agent=agent,
    eval_dataset_file_path="eval/email_agent_test_cases.json",
)

Use a test API key during ADK evaluation runs. Test-mode calls to send_email record the send without delivering the message, so full agent sessions can be evaluated without side effects.

Getting started

Step by step

1

Install Google ADK and create a MultiMail account

Install the ADK package with pip and sign up at multimail.dev to get your API key and provision a mailbox. Use a test key (mm_test_...) during development.

bash
pip install google-adk requests

"cm"># Set your MultiMail API key as an environment variable
export MULTIMAIL_API_KEY="mm_test_your_test_key_here"
2

Wrap MultiMail endpoints as typed tool functions

Define Python functions with type annotations and docstrings for each MultiMail operation your agent needs. ADK builds the tool schema from the signatures automatically — no additional registration is required.

bash
import os
import requests

MULTIMAIL_API = "https://api.multimail.dev"
HEADERS = {
    "Authorization": f"Bearer {os.environ[&"cm">#039;MULTIMAIL_API_KEY']}",
    "Content-Type": "application/json",
}


def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email from the agent&"cm">#039;s mailbox. Returns message_id and delivery status."""
    resp = requests.post(
        f"{MULTIMAIL_API}/send_email",
        headers=HEADERS,
        json={"to": to, "subject": subject, "body": body},
    )
    resp.raise_for_status()
    return resp.json()


def check_inbox(mailbox: str, unread_only: bool = True) -> dict:
    """List recent emails. Returns a list of message summaries with stable IDs."""
    resp = requests.get(
        f"{MULTIMAIL_API}/check_inbox",
        headers=HEADERS,
        params={"mailbox": mailbox, "unread_only": unread_only},
    )
    resp.raise_for_status()
    return resp.json()
3

Create the ADK agent with MultiMail tools

Pass the tool functions to the Agent constructor. Choose an oversight mode in the MultiMail dashboard for your mailbox — gated_send holds every outbound message for human review before delivery.

bash
from google.adk.agents import Agent

agent = Agent(
    model="gemini-2.0-flash",
    name="email_agent",
    instruction=(
        "You are an email assistant. Use check_inbox to read incoming messages "
        "and send_email to respond. Always confirm the recipient before sending."
    ),
    tools=[send_email, check_inbox],
)
4

Run the agent session

Use ADK's Runner and InMemorySessionService to execute the agent. Test keys ensure no real emails are sent during development. Swap to mm_live_... when ready for production.

bash
import asyncio
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types


async def main():
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="email_agent", user_id="dev"
    )
    runner = Runner(
        agent=agent,
        app_name="email_agent",
        session_service=session_service,
    )
    async for event in runner.run_async(
        user_id="dev",
        session_id=session.id,
        new_message=types.Content(
            role="user",
            parts=[types.Part(text="Check my inbox and summarize any unread messages.")],
        ),
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)


asyncio.run(main())
FAQ

Common questions

How does MultiMail's oversight work with ADK's async session model?
ADK sessions are async — the agent continues processing while waiting for external results. When gated_send mode holds a message, send_email returns immediately with status pending_approval and does not block the session. MultiMail delivers the message only after a human calls decide_email (via webhook or direct API call). The agent can proceed to other work within the same session without waiting for the approval.
Can I give each ADK agent its own email identity?
Yes. Create a separate MultiMail mailbox for each agent — for example, [email protected] and [email protected]. Each mailbox has its own API key and oversight mode. Recipients see a distinct sender address per agent, and you can apply different oversight settings to each mailbox independently.
Does MultiMail work with ADK's evaluation framework?
Yes. Pass a test API key (mm_test_...) when running evaluation workflows. Test-mode calls to send_email record the request and return a valid response without delivering the email to the recipient. Switch to an mm_live_... key for production. The tool function signatures are identical in both modes — no code changes are needed.
How do I handle inbound email in an ADK session?
MultiMail fires a webhook when inbound email arrives. Use this webhook to trigger a new ADK session or inject a message into an existing one. Alternatively, call check_inbox at the start of each session to fetch unread messages, then use read_email to retrieve the full body. The message_id returned by check_inbox is stable and can be passed directly to reply_email to respond in the same thread.
Can I use MultiMail in a multi-agent ADK setup?
Yes. In a multi-agent ADK architecture, each sub-agent can receive a different subset of MultiMail tools. For example, a triage agent might only have check_inbox and tag_email, while a response agent has reply_email and send_email. Each agent's mailbox can have a different oversight mode, so higher-trust agents can operate in monitored mode while lower-trust agents use gated_send.
What oversight mode should I start with for a new ADK agent?
Start with gated_send. It lets the agent operate autonomously for reads and drafts while requiring human approval for every outbound send. Once you've reviewed the agent's send behavior over a representative sample of sessions and are confident in its judgment, you can move to monitored (agent sends freely, human receives notifications) or autonomous (fully unattended). This graduated approach matches ADK's evaluation-driven development model.
Related integrations

Explore more

LangGraph

orchestration · both

View guide →

OpenAI Agents SDK

orchestration · python

View guide →

Semantic Kernel

orchestration · both

View guide →
Get started

The only agent email with a verifiable sender

Email infrastructure built for AI agents. Verifiable identity, graduated oversight, and a 38-tool MCP server. Formally verified in Lean 4.

Read the Docs →See Pricing