Use ControlFlow's task-based workflow orchestration to build reliable, multi-step email processing pipelines with MultiMail's delivery infrastructure and human oversight.
ControlFlow is a workflow orchestration framework by Prefect that treats AI tasks as first-class workflow steps. It combines Prefect's workflow reliability with LLM-powered task execution and structured outputs. MultiMail provides the email infrastructure that ControlFlow workflows need to send, receive, and manage messages as part of larger business processes.
By integrating MultiMail with ControlFlow, you can model email tasks as explicit workflow steps with dependencies, structured results, and error handling. Email approval in MultiMail's gated_send mode becomes a workflow dependency that must resolve before downstream tasks proceed — making approval a first-class workflow primitive.
Connect ControlFlow to MultiMail by defining tasks that call the REST API. ControlFlow's dependency system ensures email operations execute in the correct order with proper error handling and retry logic.
ControlFlow's task dependency system can model email approval as an explicit step. MultiMail's pending queue becomes a dependency that must resolve before the workflow continues, making oversight a natural part of the flow.
ControlFlow tasks produce typed, structured results. Define result types for email operations — classification outcomes, delivery confirmations, thread summaries — and pass them between workflow steps with type safety.
Built on Prefect's workflow engine, ControlFlow provides retry logic, error handling, and observability for email workflows. Failed API calls retry automatically, and you can monitor workflow health through Prefect's dashboard.
Chain email tasks with dependencies: classify inbound mail, route to appropriate handlers, compose responses, approve, and send. ControlFlow ensures each step executes only when its dependencies are satisfied.
Assign different email tasks to specialized ControlFlow agents. A classification agent triages, a composition agent drafts, and a review agent validates — all coordinated through the workflow engine.
import controlflow as cf
import requests
from pydantic import BaseModel
MULTIMAIL_API = "https://api.multimail.dev/v1"
HEADERS = {"Authorization": "Bearer mm_live_your_api_key"}
class EmailSummary(BaseModel):
message_id: str
sender: str
subject: str
category: str
urgency: str
class ReplyDraft(BaseModel):
message_id: str
body: str
tone: str
@cf.flow
def email_processing_flow(mailbox_id: str):
"cm"># Task 1: Fetch and classify inbox
summaries = cf.Task(
"Fetch emails from the inbox and classify each by category and urgency",
result_type=list[EmailSummary],
tools=[fetch_inbox_tool(mailbox_id)]
)
"cm"># Task 2: Draft replies (depends on classification)
drafts = cf.Task(
"Draft professional replies for urgent and routine emails",
result_type=list[ReplyDraft],
depends_on=[summaries],
context={"summaries": summaries}
)
"cm"># Task 3: Send replies via MultiMail
cf.Task(
"Send the drafted replies through MultiMail",
depends_on=[drafts],
tools=[send_reply_tool()],
context={"drafts": drafts}
)
return draftsCreate ControlFlow tasks for email operations backed by MultiMail's API.
def fetch_inbox_tool(mailbox_id: str):
def fetch_inbox() -> str:
"""Fetch recent emails from the MultiMail inbox."""
resp = requests.get(
f"{MULTIMAIL_API}/mailboxes/{mailbox_id}/inbox",
headers=HEADERS, params={"limit": 20}
)
return str(resp.json())
return fetch_inbox
def send_reply_tool():
def send_reply(message_id: str, body: str) -> str:
"""Reply to an email via MultiMail. In gated_send mode, queues for approval."""
resp = requests.post(f"{MULTIMAIL_API}/reply", headers=HEADERS, json={
"message_id": message_id,
"body": body
})
return str(resp.json())
return send_reply
def send_email_tool():
def send_email(to: str, subject: str, body: str, mailbox_id: str) -> str:
"""Send a new email via MultiMail. In gated_send mode, queues for approval."""
resp = requests.post(f"{MULTIMAIL_API}/send", headers=HEADERS, json={
"mailbox_id": mailbox_id, "to": to,
"subject": subject, "body": body
})
return str(resp.json())
return send_emailDefine tool functions that ControlFlow agents can use to interact with MultiMail.
triage_agent = cf.Agent(
name="Triage",
instructions="You classify emails by category and urgency. "
"Categories: support, sales, billing, spam. "
"Urgency: high, medium, low."
)
composer_agent = cf.Agent(
name="Composer",
instructions="You draft professional email replies. "
"Be concise, helpful, and match the appropriate tone. "
"The mailbox uses gated_send mode, so replies "
"will be reviewed before delivery."
)
@cf.flow
def multi_agent_email_flow(mailbox_id: str):
"cm"># Triage agent classifies emails
classifications = cf.Task(
"Classify all inbox emails by category and urgency",
result_type=list[EmailSummary],
agents=[triage_agent],
tools=[fetch_inbox_tool(mailbox_id)]
)
"cm"># Composer agent drafts replies for non-spam
drafts = cf.Task(
"Draft replies for all non-spam emails. Match tone to urgency.",
result_type=list[ReplyDraft],
agents=[composer_agent],
depends_on=[classifications],
context={"classifications": classifications}
)
"cm"># Send all drafts via MultiMail
cf.Task(
"Send all drafted replies through MultiMail",
depends_on=[drafts],
tools=[send_reply_tool()],
context={"drafts": drafts}
)
return drafts
result = multi_agent_email_flow("your_mailbox_id")Use specialized ControlFlow agents for different email processing roles.
Sign up at multimail.dev, create a mailbox, and generate an API key. Your key will start with mm_live_.
Install ControlFlow and requests for calling the MultiMail API.
pip install controlflow requestsCreate Python functions that wrap MultiMail API endpoints for fetching inbox, sending email, and replying to threads.
Define a ControlFlow @flow with tasks that have explicit dependencies. Assign agents and tools to each task.
@cf.flow
def email_flow(mailbox_id: str):
classify = cf.Task("Classify inbox", result_type=list[EmailSummary])
reply = cf.Task("Draft replies", depends_on=[classify])Execute the flow. Monitor execution in Prefect's dashboard and review pending emails in MultiMail's dashboard.
result = email_flow("your_mailbox_id")Email infrastructure built for AI agents. Verifiable identity, graduated oversight, and a 38-tool MCP server. Formally verified in Lean 4.