LangGraph·Py SDK·6 min read

Mails.ai with LangGraph — function tools for orchestrated agent flows

langchain-ai.github.io/langgraph/

Py SDK

Install with `pip install langgraph mailsai`. Then wrap the mails REST API as function tools and bind to your StateGraph nodes:

from langgraph.graph import StateGraph, END
from langchain_core.tools import tool
from langchain_anthropic import ChatAnthropic
from mailsai import agent
import os

# 1. Wire mails.ai
sarah = agent("sarah", api_key=os.environ["MAILS_API_KEY"])

@tool
def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email from the sarah agent."""
    result = sarah.send(to=to, subject=subject, body=body)
    return {"send_id": result.id, "pool": result.pool}

@tool
def list_recent_threads(since_days: int = 1) -> list:
    """List recent inbound threads on the sarah agent."""
    return [t.dict() for t in sarah.list_threads(since_days=since_days)]

# 2. Bind to a model node
model = ChatAnthropic(model="claude-sonnet-4-6").bind_tools([send_email, list_recent_threads])

# 3. Wire into a StateGraph
graph = StateGraph(YourStateType)
graph.add_node("agent", lambda s: {"messages": [model.invoke(s["messages"])]})
graph.add_node("tools", ToolNode([send_email, list_recent_threads]))
graph.add_edge("agent", "tools")
graph.add_edge("tools", END)

LangGraph is the most-used Python orchestration framework for agent workflows — StateGraph for explicit state machines, conditional edges for branching logic, persistent checkpointing for long-running flows. Wire mails.ai as function tools and any node in your graph can send + read email as part of the workflow.

Why LangGraph + mails.ai

LangGraph’s explicit-state-machine model maps naturally onto agent-mail workflows where the state of the conversation matters across many steps:

Sales outreach pipelines. StateGraph nodes for find_lead → research → draft_email → send → wait_for_reply → classify_response→ conditional branches based on intent. Persistent state so each lead progresses independently.
Support escalation flows.Inbound typed event triggers a graph; the state machine walks through triage → auto-reply → (escalate-to-human if urgency > 0.8) → resolve. The whole conversation lifecycle is graph state.
Lifecycle automations driven by an agent. Onboarding sequences where each step depends on what the user did or said in the previous email. State machine + persistent checkpointing means the user can take days between replies without state loss.

The two integration paths

Path A — @tool function wrappers (recommended). The install card pattern above. Idiomatic LangGraph code, no extra dependencies, full type hints for the state graph.

Path B — MCP server via langchain-mcp-adapters. If your team already uses MCP across other runtimes (Claude Code, Cursor, OpenAI Agents SDK) and wants consistency:

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent

async with MultiServerMCPClient(
    {
        "mails": {
            "command": "npx",
            "args": ["-y", "@mailsai/mcp-server"],
            "env": {"MAILS_API_KEY": os.environ["MAILS_API_KEY"]},
            "transport": "stdio",
        },
    }
) as client:
    agent_runnable = create_react_agent(model, client.get_tools())
    result = await agent_runnable.ainvoke(
        {"messages": [HumanMessage(content="Email lead@example.com a follow-up")]}
    )

Path B trades a dependency for MCP-consistency across your stack. Path A is simpler for teams using LangGraph as their primary framework.

Inbound handling patterns

# Pattern: webhook -> StateGraph
from fastapi import FastAPI, Request
from langgraph.checkpoint.postgres import PostgresSaver

app = FastAPI()
checkpointer = PostgresSaver.from_conn_string(POSTGRES_URI)
graph = your_support_graph.compile(checkpointer=checkpointer)

@app.post("/webhooks/mails-inbound")
async def handle_inbound(req: Request):
    event = await req.json()

    # Inject the typed event into graph state
    config = {"configurable": {"thread_id": event["thread_id"]}}
    await graph.ainvoke(
        {
            "messages": [
                {"role": "user", "content": f"Inbound from {event['from']}: {event['body']}"},
            ],
            "current_event": event,  # full typed event with intent, entities, injection_score
        },
        config=config,
    )
    return {"ok": True}

Common patterns

State-resumable email loops.The Postgres checkpointer persists conversation state. If your agent emails a lead and waits 3 days for a reply, the graph state is preserved — the next inbound webhook resumes the graph from the waiting state.
Parallel fan-out for outreach.Use LangGraph’s parallel-edge patterns to email N leads simultaneously. Idempotency keys on send_email prevent duplicates if a node retries.
Conditional branching on intent. The typed event’s intentfield becomes the branching condition for routing to different state graph paths (refund_request → refund-handling subgraph, schedule_demo → calendar-booking subgraph).

Security considerations

Injection guard before state injection. Webhook handler should check event["injection_score"] > 0.5before injecting the event into graph state. Once it’s in state, the agent will reason against it.
Per-graph mails.ai key. Different StateGraphs (different agent identities) get different mails.ai keys. Key rotation per-graph without affecting other workflows.
Send budget enforcement at the graph layer. StateGraph state can track per-thread send count and conditionally short-circuit before exceeding a per-thread limit. Belt-and-suspenders to platform-side limits (Phase 1.5).

Compare against the Anthropic SDK setup for a more direct (non-orchestration) integration, or Pydantic AI for typed-input agent definitions.

FAQ

Questions developers ask after wiring this up.

Why function tools instead of MCP for LangGraph?

LangGraph doesn't have first-class MCP-client primitives in its core API today. You can use the langchain-mcp-adapters package to expose MCP servers as LangChain tools, but it adds a dependency layer and the @tool function approach is more idiomatic LangGraph code. If your team prefers MCP for consistency with other runtimes, the adapter works — just add `pip install langchain-mcp-adapters` and wrap the mails MCP server.

How do I handle inbound webhooks in a LangGraph workflow?

Two patterns. (1) Polling: a scheduled LangGraph run that calls list_recent_threads, processes new typed events, and updates state. Good for low-volume async flows. (2) Webhook-driven: a separate FastAPI endpoint receives the webhook from mails.ai, persists the event to your StateGraph's checkpointer, and triggers the relevant graph from the right state. The second pattern matches how LangGraph customers usually wire external triggers.

Does this work with LangGraph's persistent state / checkpointing?

Yes. mails.send results (send_id, pool, classifier_score) can be stored in your StateGraph's state — useful for resumable flows where the graph picks up after an email is sent and waits for a reply. The checkpointer (PostgresSaver, SqliteSaver, etc.) persists the full state graph including any mails.* tool call results.

Can I run mails.* tools in parallel LangGraph nodes?

Yes. mails.send is idempotent when you supply a client-side request_id (the SDK does this automatically). Parallel fan-out (e.g., emailing 100 leads simultaneously via parallel branches in StateGraph) works without duplicate-send risk. Per-agent rate limits still apply — see the Limits doc at Phase 1 launch.