class EmailAssistant(Agent):
inbox_id: str
ws_task: asyncio.Task | None = None
def __init__(self) -> None:
client = AgentMail()
# By setting the client_id the inbox is created only once.
username = os.getenv("AGENTMAIL_USERNAME")
inbox = client.inboxes.create(username=username, client_id=f"{username}-inbox")
self.inbox_id = inbox.inbox_id
super().__init__(
instructions=f"""
You are a helpful voice and email AI assistant. Your name is AgentMail. You can receive emails at {self.inbox_id}. You can also send and reply to emails.
When using email tools, use "{self.inbox_id}" as the inbox_id parameter. When writing emails, include "AgentMail" in the signature.
Always speak in English.
IMPORTANT: {self.inbox_id} is your inbox, not the user's inbox.
""",
# The AgentMail Toolkit has ready-to-go tools for LiveKit agents.
tools=AgentMailToolkit(client=client).get_tools(
[
"list_threads",
"get_thread",
"get_attachment",
"send_message",
"reply_to_message",
]
),
)
async def _websocket_task(self):
# Open a websocket connection to AgentMail.
async with AsyncAgentMail().websockets.connect() as socket:
# Subscribe to events from the inbox.
await socket.send_subscribe(Subscribe(inbox_ids=[self.inbox_id]))
while True:
data = await socket.recv()
# If a message is received by the inbox, interrupt the current conversation and generate a reply.
if isinstance(data, MessageReceived):
self.session.interrupt()
await self.session.generate_reply(
instructions=f"""Say "I've received an email" and then read the email.""",
user_input=data.message.model_dump_json(),
)
# Open the websocket connection and generate a greeting when the agent enters the call.
async def on_enter(self):
self.ws_task = asyncio.create_task(self._websocket_task())
await self.session.generate_reply(
instructions=f"""In English, greet the user, introduce yourself as AgentMail, inform them that you can "receive emails" at {self.inbox_id}, and offer your assistance.""",
allow_interruptions=False,
)
# Close the websocket connection when the agent exits the call.
async def on_exit(self):
if self.ws_task:
self.ws_task.cancel()