CarlyEmail docs

LangChain and LangGraph

Load CarlyEmail's hosted MCP tools, invoke a LangGraph agent from verified mail, and preserve one graph thread per email thread.

The simplest LangChain MCP integration uses CarlyEmail's hosted server. It supplies the live tool schemas, while a FastAPI webhook invokes the graph whenever a real email arrives.

1. Install and create an inbox

pip install langchain langchain-openai langchain-mcp-adapters fastapi svix uvicorn
npx carlyemail signup --human-email you@example.com --username assistant
npx carlyemail verify 123456

export OPENAI_API_KEY=sk-...
export CARLYEMAIL_API_KEY=ce_us_...
export CARLYEMAIL_INBOX=assistant@carlyemail.com
export CARLYEMAIL_WEBHOOK_SECRET=whsec_...

2. Load CarlyEmail's tools

Python

import asyncio
import os

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient


async def main():
    client = MultiServerMCPClient({
        "carlyemail": {
            "transport": "http",
            "url": "https://api.carlyemail.com/mcp",
            "headers": {
                "Authorization": f"Bearer {os.environ['CARLYEMAIL_API_KEY']}"
            },
        }
    })
    tools = await client.get_tools()
    agent = create_agent(
        model="openai:gpt-5-mini",
        tools=tools,
        system_prompt=(
            f"You manage {os.environ['CARLYEMAIL_INBOX']}. "
            "Read the message and thread before acting. "
            "Draft rather than send when intent or authority is ambiguous."
        ),
    )
    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": "Summarize new mail."}]
    })
    print(result["messages"][-1].content)


asyncio.run(main())

MCP removes the wrapper-maintenance step. Filter tools before passing them to create_agent when the model needs only a smaller surface, and enforce the allowed operations with an inbox-scoped API key.

3. Make incoming email invoke the graph

This complete FastAPI app keeps a graph checkpoint thread for each CarlyEmail thread_id:

Python

import json
import os
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, Response
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.memory import InMemorySaver
from svix.webhooks import Webhook, WebhookVerificationError


@asynccontextmanager
async def lifespan(app: FastAPI):
    client = MultiServerMCPClient({
        "carlyemail": {
            "transport": "http",
            "url": "https://api.carlyemail.com/mcp",
            "headers": {
                "Authorization": f"Bearer {os.environ['CARLYEMAIL_API_KEY']}"
            },
        }
    })
    app.state.agent = create_agent(
        model="openai:gpt-5-mini",
        tools=await client.get_tools(),
        checkpointer=InMemorySaver(),
        system_prompt=(
            "You manage an email inbox. Read the thread before acting. "
            "Reply in the existing thread, and draft when unsure."
        ),
    )
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/hooks/carlyemail")
async def on_email(request: Request):
    raw = await request.body()
    try:
        event = Webhook(os.environ["CARLYEMAIL_WEBHOOK_SECRET"]).verify(
            raw, request.headers
        )
    except WebhookVerificationError:
        return Response("Invalid signature", status_code=400)

    message = event.get("message")
    if event.get("event_type") != "message.received" or not message:
        return Response(status_code=204)

    await request.app.state.agent.ainvoke(
        {
            "messages": [{
                "role": "user",
                "content": (
                    "Handle this newly received email. Use CarlyEmail's reply "
                    "tool if a reply is appropriate.\n\n" + json.dumps(message)
                ),
            }]
        },
        {"configurable": {"thread_id": message["thread_id"]}},
    )
    return Response(status_code=204)

InMemorySaver makes the example runnable. Replace it with a durable LangGraph checkpointer in production, keeping CarlyEmail thread_id as the graph thread_id.

Run and register the route:

uvicorn app:app --host 0.0.0.0 --port 8000
npx carlyemail webhook https://your-agent.example/hooks/carlyemail \
  --events message.received

Optional first-party package

The repository also contains langchain-carlyemail: typed tools, a message loader, a search retriever, and a verified inbound router. Until its first PyPI release, install it from Git:

pip install \
  "langchain-carlyemail[webhooks] @ git+https://github.com/shirschfield/carlyemail.git@main#subdirectory=integrations/langchain-carlyemail"

Python

from langchain_carlyemail import CarlyEmailLoader, CarlyEmailRetriever, CarlyEmailToolkit

tools = CarlyEmailToolkit.from_api_key().get_tools()
documents = CarlyEmailLoader("assistant@carlyemail.com", max_messages=200).load()
retriever = CarlyEmailRetriever.from_api_key("assistant@carlyemail.com", limit=10)

The package's send and reply tool descriptions explicitly say that sending is immediate and cannot be recalled. Its create_email_router helper performs the same raw-body verification shown above.

Warning

Email bodies and attachments are untrusted model input. Deduplicate by event_id, scope the key to one inbox, and omit message_send when a person should approve drafts before they leave.

See also