Email infrastructure for Agno agents and teams

import requests
from agno.agent import Agent
from agno.models.openai import OpenAIChat

MULTIMAIL_API_KEY = "mm_live_your_key_here"
MULTIMAIL_BASE = "https://api.multimail.dev"

def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email via MultiMail. Returns message_id and status."""
    resp = requests.post(
        f"{MULTIMAIL_BASE}/send_email",
        headers={"Authorization": f"Bearer {MULTIMAIL_API_KEY}"},
        json={
            "from": "[email protected]",
            "to": to,
            "subject": subject,
            "body": body,
        },
    )
    resp.raise_for_status()
    return resp.json()

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[send_email],
    description="You are an outreach agent. Use send_email when you need to contact someone.",
)

agent.print_response(
    "Send a follow-up email to [email protected] about the Q2 integration demo."
)

import requests
from agno.agent import Agent
from agno.models.openai import OpenAIChat

MULTIMAIL_API_KEY = "mm_live_your_key_here"
MULTIMAIL_BASE = "https://api.multimail.dev"
HEADERS = {"Authorization": f"Bearer {MULTIMAIL_API_KEY}"}

def check_inbox(mailbox: str, limit: int = 10) -> list:
    """Return recent messages for a mailbox."""
    resp = requests.get(
        f"{MULTIMAIL_BASE}/check_inbox",
        headers=HEADERS,
        params={"mailbox": mailbox, "limit": limit},
    )
    resp.raise_for_status()
    return resp.json()["messages"]

def get_thread(thread_id: str) -> list:
    """Return all messages in a thread, oldest first."""
    resp = requests.get(
        f"{MULTIMAIL_BASE}/get_thread/{thread_id}",
        headers=HEADERS,
    )
    resp.raise_for_status()
    return resp.json()["messages"]

def reply_email(message_id: str, body: str) -> dict:
    """Reply to an existing message, preserving the thread."""
    resp = requests.post(
        f"{MULTIMAIL_BASE}/reply_email",
        headers=HEADERS,
        json={"message_id": message_id, "body": body},
    )
    resp.raise_for_status()
    return resp.json()

support_agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[check_inbox, get_thread, reply_email],
    description="You are a support agent. Check the inbox, read the full thread before replying, and keep replies concise.",
)

support_agent.print_response(
    "Check [email protected] for any open questions and reply to each one."
)

import requests
from agno.agent import Agent
from agno.team import Team
from agno.models.openai import OpenAIChat

MULTIMAIL_API_KEY = "mm_live_your_key_here"
MULTIMAIL_BASE = "https://api.multimail.dev"
HEADERS = {"Authorization": f"Bearer {MULTIMAIL_API_KEY}"}

"cm"># Mailbox is configured with oversight_mode=gated_send in the MultiMail dashboard.
"cm"># Calls to send_email will queue the message for human approval, not deliver it immediately.

def send_email(to: str, subject: str, body: str) -> dict:
    resp = requests.post(
        f"{MULTIMAIL_BASE}/send_email",
        headers=HEADERS,
        json={"from": "[email protected]", "to": to, "subject": subject, "body": body},
    )
    resp.raise_for_status()
    return resp.json()  "cm"># {"message_id": "...", "status": "pending_approval"}

def list_pending() -> list:
    """Return messages waiting for human approval."""
    resp = requests.get(f"{MULTIMAIL_BASE}/list_pending", headers=HEADERS)
    resp.raise_for_status()
    return resp.json()["messages"]

research_agent = Agent(
    name="Researcher",
    model=OpenAIChat(id="gpt-4o"),
    description="You gather information and summarize findings for the communications agent.",
)

comms_agent = Agent(
    name="Communications",
    model=OpenAIChat(id="gpt-4o"),
    tools=[send_email, list_pending],
    description="You draft and queue outbound emails based on the researcher&"cm">#039;s findings. Always call list_pending after sending to confirm the message is in the approval queue.",
)

team = Team(
    name="OutreachTeam",
    agents=[research_agent, comms_agent],
    model=OpenAIChat(id="gpt-4o"),
)

team.print_response(
    "Research the top 3 open-source Python agent frameworks by GitHub stars and draft a partnership inquiry email to each project&"cm">#039;s maintainers."
)

import requests
from fastapi import FastAPI, Request
from agno.agent import Agent
from agno.models.openai import OpenAIChat

app = FastAPI()

MULTIMAIL_API_KEY = "mm_live_your_key_here"
MULTIMAIL_BASE = "https://api.multimail.dev"
HEADERS = {"Authorization": f"Bearer {MULTIMAIL_API_KEY}"}

def tag_email(message_id: str, tag: str) -> dict:
    resp = requests.post(
        f"{MULTIMAIL_BASE}/tag_email",
        headers=HEADERS,
        json={"message_id": message_id, "tag": tag},
    )
    resp.raise_for_status()
    return resp.json()

def reply_email(message_id: str, body: str) -> dict:
    resp = requests.post(
        f"{MULTIMAIL_BASE}/reply_email",
        headers=HEADERS,
        json={"message_id": message_id, "body": body},
    )
    resp.raise_for_status()
    return resp.json()

triage_agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[tag_email, reply_email],
    description=(
        "You triage inbound emails. Tag urgent messages as &"cm">#039;urgent', tag billing questions as 'billing', "
        "and reply to simple questions immediately. For anything complex, tag as &"cm">#039;needs-human' and do not reply."
    ),
)

@app.post("/webhooks/inbound")
async def handle_inbound(request: Request):
    payload = await request.json()
    message_id = payload["message_id"]
    sender = payload["from"]
    subject = payload["subject"]
    body = payload["body_text"]
    dkim_valid = payload.get("dkim_valid", False)

    if not dkim_valid:
        # Do not act on messages that fail DKIM — tag for human review.
        tag_email(message_id, "dkim-failed")
        return {"status": "held"}

    triage_agent.run(
        f"New email from {sender}.\nSubject: {subject}\nBody: {body}\nMessage ID: {message_id}\n\nTriage and respond as appropriate."
    )
    return {"status": "processed"}