Send & Receive Emails with LiveKit Agents

This page gives your LiveKit voice agent its own @agentmail.to inbox. Ask it to send an email and it sends one. Email it mid-call and it interrupts itself to read the message aloud. You will extend the agent.py you built in LiveKit’s Voice AI quickstart, in Python.

0. Prerequisites

You need three things:

• An AgentMail API key. Generate one from the AgentMail Console, or follow the Quickstart if this is your first AgentMail project.
• A working voice agent from the LiveKit Voice AI quickstart. This page modifies its agent.py.
• Python 3.11 or newer.

Install the AgentMail packages into the same environment as your LiveKit project:

pip install agentmail agentmail-toolkit

Set your API key and pick a username for your agent’s inbox:

export AGENTMAIL_API_KEY="<API_KEY>"
export AGENTMAIL_USERNAME="<username for your agent's inbox>"

Both variables can also go in the .env.local file your quickstart project already loads.

1. Create the agent’s inbox

An inbox is an email address your agent owns. The agent creates its own when it starts, so there is no provisioning step to run by hand.

Add the imports to the top of agent.py:

import os
import asyncio

from agentmail import AgentMail, AsyncAgentMail, Subscribe, MessageReceivedEvent
from agentmail.inboxes import CreateInboxRequest
from agentmail_toolkit.livekit import AgentMailToolkit

This is the create call. It goes inside the agent class in the next step:

client = AgentMail()

username = os.environ["AGENTMAIL_USERNAME"]
inbox = client.inboxes.create(
    request=CreateInboxRequest(username=username, client_id=f"{username}-livekit")
)

The client_id makes the create idempotent. The first startup creates <username>@agentmail.to. Every later startup with the same client_id returns that same inbox instead of failing, so your agent keeps one address across restarts.

Usernames are claimed once, for everyone on AgentMail.

2. Give the agent email tools

Replace the quickstart’s Assistant class with an EmailAssistant. It creates the inbox from the previous step, tells the model the address is its own, and passes in email tools from the AgentMail toolkit. Agent is already imported in your quickstart file.

class EmailAssistant(Agent):
    def __init__(self) -> None:
        client = AgentMail()

        username = os.environ["AGENTMAIL_USERNAME"]
        inbox = client.inboxes.create(
            request=CreateInboxRequest(username=username, client_id=f"{username}-livekit")
        )
        self.inbox_id = inbox.inbox_id
        self.ws_task: asyncio.Task | None = None

        super().__init__(
            instructions=f"""
            You are a helpful voice assistant with your own email inbox.
            Your address is {inbox.email}. It is your inbox, not the user's.
            Pass "{self.inbox_id}" as the inbox_id parameter of every email tool.
            When you greet the user, tell them your email address.
            """,
            tools=AgentMailToolkit(client=client).get_tools(
                [
                    "list_threads",
                    "get_thread",
                    "get_attachment",
                    "send_message",
                    "reply_to_message",
                ]
            ),
        )

The instructions carry two values from the create response: the inbox’s email, the address the agent tells the user, and its inbox_id, which the model passes to every tool instead of guessing.

get_tools takes the names of the tools you want, and calling it with no argument returns all of them:

Two behaviors come built into the toolkit:

• While a tool call runs, the agent speaks a short update about what it is doing, so the user is not left waiting in silence.
• When a call fails, the API’s error message goes back to the model as the tool result, so the agent can explain the problem or try again.

3. Test sending emails by voice

Point the session at your new agent. In the function that starts the session, replace Assistant() with EmailAssistant() and keep every other option your file already passes:

await session.start(
    room=ctx.room,
    agent=EmailAssistant(),
)

Run the agent in your terminal:

python agent.py console

Ask it out loud to send an email to your personal address with a short message. The agent tells you it is sending while the tool call runs, and a moment later the email shows up in your account, from <username>@agentmail.to.

Spoken addresses are easy to mishear. Have the agent repeat the address back before it sends, or put your address in the instructions so the model does not have to transcribe it.

4. Test receiving emails mid-call

Sending happens when the model decides to call a tool. Receiving needs a push in the other direction. Your agent keeps a WebSocket connection to AgentMail open for the whole session and gets a message.received event the moment mail arrives:

Add three methods to EmailAssistant. They use the ws_task attribute the constructor already set:

    async def _watch_inbox(self) -> None:
        client = AsyncAgentMail()

        async with client.websockets.connect() as socket:
            await socket.send_subscribe(
                Subscribe(inbox_ids=[self.inbox_id], event_types=["message.received"])
            )

            async for event in socket:
                if isinstance(event, MessageReceivedEvent):
                    self.session.interrupt()
                    await self.session.generate_reply(
                        instructions='Say "I just received an email", then read it to the user.',
                        user_input=event.message.model_dump_json(),
                    )

    async def on_enter(self) -> None:
        self.ws_task = asyncio.create_task(self._watch_inbox())

    async def on_exit(self) -> None:
        if self.ws_task:
            self.ws_task.cancel()

How the pieces fit together:

• LiveKit calls on_enter when your agent enters the session and on_exit when it leaves, so the connection lives exactly as long as the conversation.
• The subscribe frame filters to message.received, so the loop wakes only for inbound mail.
• The event carries the full message, body included. session.interrupt() stops whatever the agent is saying, and generate_reply feeds the email in as input, so the agent reads it in its own voice.

Run python agent.py console again. Mid-conversation, send an email from your regular account to your agent’s address. Within a few seconds the agent cuts itself off, announces the email, and reads it. Ask it to reply, and the reply_to_message tool answers in the same thread.

Events are delivered only while the connection is open. For reconnecting after a drop and the other events you can subscribe to, see WebSockets.

Next Steps