Email handoff for LiveKit voice agents

import httpx
from livekit.agents import AgentSession, JobContext, WorkerOptions, cli
from livekit.agents.llm import ChatContext

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

async def send_session_summary(
    recipient_email: str,
    recipient_name: str,
    summary: str,
    session_id: str,
) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MULTIMAIL_BASE}/send_email",
            headers={"Authorization": f"Bearer {MULTIMAIL_API_KEY}"},
            json={
                "from": "[email protected]",
                "to": recipient_email,
                "subject": "Summary from your session",
                "body": f"Hi {recipient_name},\n\nHere&"cm">#039;s a summary of our conversation:\n\n{summary}\n\nReply to this email if you have questions.",
                "oversight_mode": "gated_send",
                "metadata": {"session_id": session_id},
            },
        )
        response.raise_for_status()
        return response.json()

async def entrypoint(ctx: JobContext):
    await ctx.connect()

    session = AgentSession()

    @session.on("session_ended")
    async def on_session_ended(summary_text: str):
        # Extract contact info from your session metadata
        participant = ctx.room.remote_participants.get(ctx.room.name)
        if participant:
            email = participant.metadata  # store email in participant metadata
            await send_session_summary(
                recipient_email=email,
                recipient_name=participant.name,
                summary=summary_text,
                session_id=ctx.room.name,
            )

    await session.start(ctx.room)

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

import httpx
from livekit.agents import AgentSession, JobContext, WorkerOptions, cli
from livekit.agents.llm import FunctionTool
from typing import Annotated

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

async def send_followup_email(
    to: Annotated[str, "Recipient email address"],
    subject: Annotated[str, "Email subject line"],
    body: Annotated[str, "Plain text body of the follow-up email"],
) -> str:
    """Queue a follow-up email to the caller. Requires human approval before delivery."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{MULTIMAIL_BASE}/send_email",
            headers={"Authorization": f"Bearer {MULTIMAIL_API_KEY}"},
            json={
                "from": "[email protected]",
                "to": to,
                "subject": subject,
                "body": body,
                "oversight_mode": "gated_send",
            },
        )
        resp.raise_for_status()
        data = resp.json()
        return f"Email queued for approval. Message ID: {data[&"cm">#039;id']}"

async def entrypoint(ctx: JobContext):
    await ctx.connect()

    session = AgentSession(
        tools=[FunctionTool.from_callable(send_followup_email)]
    )
    await session.start(ctx.room)

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

import httpx
from livekit.agents import AgentSession, JobContext, WorkerOptions, cli

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

async def fetch_recent_emails(contact_email: str, limit: int = 5) -> list[dict]:
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{MULTIMAIL_BASE}/check_inbox",
            headers={"Authorization": f"Bearer {MULTIMAIL_API_KEY}"},
            params={
                "mailbox": "[email protected]",
                "from": contact_email,
                "limit": limit,
            },
        )
        resp.raise_for_status()
        return resp.json().get("emails", [])

async def build_email_context(contact_email: str) -> str:
    emails = await fetch_recent_emails(contact_email)
    if not emails:
        return ""
    lines = ["Recent email exchanges with this contact:"]
    for email in emails:
        lines.append(f"- [{email[&"cm">#039;date']}] {email['subject']}: {email['preview']}")
    return "\n".join(lines)

async def entrypoint(ctx: JobContext):
    await ctx.connect()

    # Get contact email from participant metadata
    participant = next(iter(ctx.room.remote_participants.values()), None)
    contact_email = participant.metadata if participant else None

    email_context = ""
    if contact_email:
        email_context = await build_email_context(contact_email)

    session = AgentSession()

    # Inject email context into the initial chat context
    if email_context:
        session.chat_ctx.append(
            role="system",
            text=email_context,
        )

    await session.start(ctx.room)

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib
import httpx

app = FastAPI()

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

def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(), payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.post("/webhooks/multimail")
async def handle_multimail_webhook(request: Request):
    payload = await request.body()
    sig = request.headers.get("X-MultiMail-Signature", "")

    if not verify_signature(payload, sig, MULTIMAIL_WEBHOOK_SECRET):
        raise HTTPException(status_code=401, detail="Invalid signature")

    event = await request.json()
    event_type = event.get("type")

    if event_type == "email.approved":
        session_id = event["metadata"].get("session_id")
        message_id = event["message_id"]
        "cm"># Trigger downstream: update CRM, log to your system, notify operator
        print(f"Email {message_id} approved for session {session_id}")

    elif event_type == "email.rejected":
        session_id = event["metadata"].get("session_id")
        message_id = event["message_id"]
        print(f"Email {message_id} rejected for session {session_id} — no delivery")

    return {"ok": True}