Agno Email Toolkit

import os
from agno.tools import Toolkit
from deadsimple import DeadSimple

class DeadSimpleEmailTools(Toolkit):
    def __init__(self, **kwargs):
        self.client = DeadSimple(os.environ["DSE_API_KEY"])
        super().__init__(
            name="deadsimple_email",
            tools=[self.send_email, self.read_messages, self.wait_for_reply],
            **kwargs,
        )

    def send_email(self, inbox_id: str, to: str, subject: str, body: str) -> str:
        """Send a plain-text email from a Dead Simple inbox."""
        result = self.client.messages.send(
            inbox_id, to=[to], subject=subject, text_body=body
        )
        return f"Sent {result.message_id}"

    def read_messages(self, inbox_id: str, limit: int = 5) -> list[dict]:
        """List the most recent messages in an inbox."""
        result = self.client.messages.list(inbox_id, limit=limit)
        return [
            {"from": m.from_email, "subject": m.subject, "snippet": m.snippet}
            for m in result.messages
        ]

    def wait_for_reply(self, inbox_id: str, from_contains: str = "") -> dict | None:
        """Block until a new inbound email arrives, then return it."""
        msg = self.client.messages.wait_for(
            inbox_id, from_contains=from_contains, timeout=120
        )
        return {"from": msg.from_email, "subject": msg.subject} if msg else None

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from deadsimple_toolkit import DeadSimpleEmailTools

agent = Agent(
    model=OpenAIChat(id="gpt-4o"),
    tools=[DeadSimpleEmailTools()],
    markdown=True,
)

agent.print_response(
    "Email vendor@example.com from inbox inb_a1b2c3 asking for the "
    "August invoice, then wait for their reply and summarize it."
)

import asyncio, os
from agno.agent import Agent
from agno.tools.mcp import MCPTools, StreamableHTTPClientParams

server_params = StreamableHTTPClientParams(
    url="https://api.deadsimple.email/mcp",
    headers={"Authorization": f"Bearer {os.environ['DSE_API_KEY']}"},
)

async def main():
    async with MCPTools(
        server_params=server_params, transport="streamable-http"
    ) as mcp_tools:
        agent = Agent(tools=[mcp_tools], markdown=True)
        await agent.aprint_response(
            "Create an inbox called 'trials', sign up for the beta at "
            "example.com, and read me the verification code when it lands."
        )

asyncio.run(main())