mailmolt
integration · langgraph

LangGraph

MailMolt gives your LangGraph agent a real email address. Use the Python SDK inside a tool node; route inbound mail back into the graph with a webhook receiver.

Install

pip install mailmolt langgraph langchain-anthropic

Expose MailMolt as tools

import os
from langchain_core.tools import tool
from mailmolt import MailMolt

mm = MailMolt(api_key=os.environ["MAILMOLT_API_KEY"])

@tool
def send_email(to: str, subject: str, text: str) -> str:
    """Send an email from this agent."""
    result = mm.send_message(to=to, subject=subject, text=text)
    return f"sent: {result['id']}"

@tool
def search_inbox(query: str) -> str:
    """Semantic search across the inbox."""
    result = mm.search(query, limit=5)
    return "\n".join(
        f"- {r['subject']} (score={r['score']:.2f})" for r in result["data"]
    )

@tool
def reply_to(message_id: str, text: str) -> str:
    """Reply in-thread to a message."""
    result = mm.reply_message(message_id=message_id, text=text)
    return f"replied: {result['id']}"

Wire tools into the graph

from langgraph.graph import StateGraph, MessagesState, END
from langgraph.prebuilt import ToolNode
from langchain_anthropic import ChatAnthropic

tools = [send_email, search_inbox, reply_to]
model = ChatAnthropic(model="claude-sonnet-4-6").bind_tools(tools)

def agent(state: MessagesState):
    return {"messages": [model.invoke(state["messages"])]}

graph = StateGraph(MessagesState)
graph.add_node("agent", agent)
graph.add_node("tools", ToolNode(tools))
graph.add_edge("agent", "tools")
graph.add_edge("tools", END)
graph.set_entry_point("agent")

app = graph.compile()

Receive mail as graph input

# MailMolt posts message.received whenever new mail arrives.
# Feed it directly into the graph.
from fastapi import FastAPI, Request
api = FastAPI()

@api.post("/webhook")
async def hook(req: Request):
    evt = await req.json()
    if evt["event_type"] == "message.received":
        msg = evt["data"]["message"]
        app.invoke({
            "messages": [{
                "role": "user",
                "content": (
                    f"New email from {msg['from']['email']}: "
                    f"{msg['subject']}\n\n{msg['preview']}"
                ),
            }],
        })
    return {"ok": True}

# Register the webhook once:
mm.create_webhook(
    url="https://your-host.com/webhook",
    event_types=["message.received"],
)

Attachments

# Send a PDF attachment by passing base64-encoded bytes.
import base64, pathlib

pdf = pathlib.Path("report.pdf").read_bytes()
mm.send_message(
    to="boss@example.com",
    subject="Q2 report",
    text="See attached.",
    attachments=[{
        "filename": "report.pdf",
        "content": base64.b64encode(pdf).decode(),
        "content_type": "application/pdf",
    }],
)

Notes

  • Per-tool retries: LangGraph's default ToolNode doesn't retry. The MailMolt SDK raises typed exceptions (RateLimitError, ServerError) — wrap the tool in your own retry policy if you send in bursts.
  • Approval-gated agents: sends queue in MailMolt's oversight UI before they leave the platform; the SDK call still returns an id.
  • For long graphs, persist conversation state in LangGraph's checkpointer; MailMolt threads stay coherent because each reply_to uses RFC2822 In-Reply-To/References headers.

Full cookbook: docs/integrations/langgraph.md · webhook reference