RAG-Powered Email Agents with LlamaIndex

import requests
from llama_index.core.tools import FunctionTool

MULTIMAIL_API = "https://api.multimail.dev/v1"
HEADERS = {"Authorization": "Bearer mm_live_your_api_key"}

def send_email(to: str, subject: str, body: str, mailbox_id: str) -> str:
    """Send an email through MultiMail. In gated_send mode, this queues for human 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())

def check_inbox(mailbox_id: str, limit: int = 10) -> str:
    """Check the inbox for recent messages."""
    resp = requests.get(
        f"{MULTIMAIL_API}/mailboxes/{mailbox_id}/inbox",
        headers=HEADERS, params={"limit": limit}
    )
    return str(resp.json())

def reply_email(message_id: str, body: str) -> str:
    """Reply to an email by message ID, maintaining thread context."""
    resp = requests.post(f"{MULTIMAIL_API}/reply", headers=HEADERS, json={
        "message_id": message_id, "body": body
    })
    return str(resp.json())

send_tool = FunctionTool.from_defaults(fn=send_email)
inbox_tool = FunctionTool.from_defaults(fn=check_inbox)
reply_tool = FunctionTool.from_defaults(fn=reply_email)

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI

"cm"># Index your knowledge base
documents = SimpleDirectoryReader("./knowledge_base").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()

"cm"># Create a query tool for context retrieval
from llama_index.core.tools import QueryEngineTool
knowledge_tool = QueryEngineTool.from_defaults(
    query_engine=query_engine,
    name="knowledge_base",
    description="Search the knowledge base for information to include in emails."
)

"cm"># Build the agent with both retrieval and email tools
llm = OpenAI(model="gpt-4o")
agent = ReActAgent.from_tools(
    tools=[knowledge_tool, send_tool, inbox_tool, reply_tool],
    llm=llm,
    verbose=True,
    system_prompt="You are an email assistant. Search the knowledge base for "
    "context before composing emails. The mailbox uses gated_send mode, so "
    "sent emails require human approval before delivery."
)

response = agent.chat("Reply to the latest email about pricing with our current rates")

from llama_index.core import Document, VectorStoreIndex

def fetch_email_history(mailbox_id: str, limit: int = 100) -> list:
    """Fetch email history from MultiMail and create LlamaIndex documents."""
    resp = requests.get(
        f"{MULTIMAIL_API}/mailboxes/{mailbox_id}/inbox",
        headers=HEADERS, params={"limit": limit}
    )
    emails = resp.json().get("emails", [])
    documents = []
    for email in emails:
        doc = Document(
            text=f"From: {email[&"cm">#039;from']}\nTo: {email['to']}\n"
                 f"Subject: {email[&"cm">#039;subject']}\n\n{email['body']}",
            metadata={
                "message_id": email["message_id"],
                "date": email["date"],
                "from": email["from"]
            }
        )
        documents.append(doc)
    return documents

# Index email history
email_docs = fetch_email_history("your_mailbox_id")
email_index = VectorStoreIndex.from_documents(email_docs)
email_query = email_index.as_query_engine()

email_context_tool = QueryEngineTool.from_defaults(
    query_engine=email_query,
    name="email_history",
    description="Search past email conversations for context and prior commitments."
)

import subprocess
import json

"cm"># The MCP server can be used alongside LlamaIndex tools
"cm"># Start the MCP server process
process = subprocess.Popen(
    ["npx", "-y", "@multimail/mcp-server"],
    env={"MULTIMAIL_API_KEY": "mm_live_your_api_key"},
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE
)

"cm"># For LlamaIndex, the recommended approach is FunctionTool wrappers
"cm"># around the REST API, as shown in the previous examples.
"cm"># The MCP server is ideal for MCP-native clients.

"cm"># Combine both approaches: MCP for discovery, FunctionTool for execution
agent = ReActAgent.from_tools(
    tools=[send_tool, inbox_tool, reply_tool, knowledge_tool],
    llm=llm,
    verbose=True
)

response = agent.chat("Check my inbox and draft replies using knowledge base context")