Complete voice assistant with Gemini Live API, Firecrawl web search, and Gmail - companion code for the tutorial

agent.py

"""
Voice assistant with Gemini Live API, Firecrawl web search, and Gmail integration
"""

import asyncio
import os
import smtplib
from email.mime.text import MIMEText

from livekit.agents import (
    Agent,
    AgentSession,
    AgentServer,
    JobContext,
    cli,
    function_tool,
)
from livekit.plugins import google
from firecrawl import Firecrawl
from imap_tools import MailBox
from dotenv import load_dotenv

load_dotenv()


@function_tool
async def web_search(query: str) -> str:
    """Search the web for current information on any topic.

    Args:
        query: The search query to look up

    Returns:
        Search results with titles and descriptions
    """
    firecrawl = Firecrawl(api_key=os.getenv("FIRECRAWL_API_KEY"))

    try:
        results = firecrawl.search(query=query, limit=5)
    except Exception as e:
        return f"Search failed: {str(e)}"

    web_results = getattr(results, "web", None) or getattr(results, "data", None) or []

    if not web_results:
        return "No results found for that query."

    formatted = []
    for item in web_results:
        title = (
            getattr(item, "title", "Untitled")
            if not isinstance(item, dict)
            else item.get("title", "Untitled")
        )
        description = (
            getattr(item, "description", "")
            if not isinstance(item, dict)
            else item.get("description", "")
        )
        url = (
            getattr(item, "url", "")
            if not isinstance(item, dict)
            else item.get("url", "")
        )
        formatted.append(f"- {title}: {description} (Source: {url})")

    return "\n".join(formatted)


@function_tool
async def read_emails(count: int = 5) -> str:
    """Read recent emails from the inbox.

    Args:
        count: Number of recent emails to fetch (default 5, max 10)

    Returns:
        A summary of recent emails with sender, subject, and preview
    """
    count = min(count, 10)
    gmail_user = os.getenv("GMAIL_ADDRESS")
    gmail_password = os.getenv("GMAIL_APP_PASSWORD")

    def _fetch():
        with MailBox("imap.gmail.com").login(gmail_user, gmail_password) as mailbox:
            emails = []
            for msg in mailbox.fetch(limit=count, reverse=True):
                body_preview = (msg.text or "")[:150].replace("\n", " ").strip()
                emails.append(
                    f"- From: {msg.from_}\n  Subject: {msg.subject}\n  Preview: {body_preview}"
                )
            return emails

    try:
        emails = await asyncio.to_thread(_fetch)
    except Exception as e:
        return f"Failed to read emails: {str(e)}"

    if not emails:
        return "No emails found in inbox."

    return f"Here are your {len(emails)} most recent emails:\n\n" + "\n\n".join(emails)


@function_tool
async def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to someone.

    Args:
        to: Recipient email address
        subject: Email subject line
        body: Email body text

    Returns:
        Confirmation that the email was sent
    """
    gmail_user = os.getenv("GMAIL_ADDRESS")
    gmail_password = os.getenv("GMAIL_APP_PASSWORD")

    msg = MIMEText(body)
    msg["Subject"] = subject
    msg["From"] = gmail_user
    msg["To"] = to

    def _send():
        with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
            server.login(gmail_user, gmail_password)
            server.send_message(msg)

    try:
        await asyncio.to_thread(_send)
    except Exception as e:
        return f"Failed to send email: {str(e)}"

    return f"Email sent to {to} with subject '{subject}'."


class ResearchAssistant(Agent):
    def __init__(self):
        super().__init__(
            instructions="""You are a helpful research assistant with access to web search and email.

Your role:
- Help users find information on any topic using web search
- Read and summarize their recent emails when asked
- Send emails on their behalf when they provide a recipient, subject, and message
- Keep responses conversational and concise since this is a voice interaction
- When you use search results, mention where the information came from
- If you don't know something and search doesn't help, say so honestly

For email:
- Before sending an email, always confirm the recipient, subject, and message with the user
- When reading emails, give a brief spoken summary rather than reading every detail
- Never send an email without the user explicitly asking you to

Style guidelines:
- Speak naturally, as if having a conversation
- Avoid long lists or complex formatting that doesn't work well in speech
- Break up information into digestible pieces
- Ask clarifying questions if a request is ambiguous""",
        )


server = AgentServer()


@server.rtc_session()
async def entrypoint(ctx: JobContext):
    await ctx.connect()

    session = AgentSession(
        llm=google.realtime.RealtimeModel(
            model="gemini-2.5-flash-native-audio-preview-12-2025",
            voice="Puck",
        ),
        tools=[web_search, read_emails, send_email],
    )

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


if __name__ == "__main__":
    cli.run_app(server)