Google ADK Integration Guide

curl -X POST https://api.molted.email/v1/me/mailboxes \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Outbound Agent",
    "emailAddress": "agent@yourdomain.com"
  }'

import os
import requests


MOLTED_API_KEY = os.environ["MOLTED_API_KEY"]
MOLTED_TENANT_ID = os.environ["MOLTED_TENANT_ID"]
MOLTED_MAILBOX_ID = os.environ["MOLTED_MAILBOX_ID"]


def send_email(to: str, subject: str, body: str, dedupe_key: str = "") -> dict:
    """Send an email to a contact through the governed mailbox.

    The email is evaluated against policy rules before delivery.
    Blocked sends return a reason and are not retried automatically.

    Args:
        to: Recipient email address.
        subject: Email subject line.
        body: Email body in plain text or HTML.
        dedupe_key: Unique key to prevent duplicate sends within the cooldown window.
            Pass a stable identifier such as "{user_id}-{event_type}".
            Defaults to recipient + subject if empty.

    Returns:
        A dict with keys: status, requestId, blockReason (if blocked).
    """
    payload = {
        "tenantId": MOLTED_TENANT_ID,
        "mailboxId": MOLTED_MAILBOX_ID,
        "recipientEmail": to,
        "templateId": "_default",
        "dedupeKey": dedupe_key or f"{to}-{subject}",
        "agentId": "google-adk-agent",
        "payload": {
            "subject": subject,
            "html": body,
            "text": body,
        },
    }

    resp = requests.post(
        "https://api.molted.email/v1/agent/send",
        json=payload,
        headers={"Authorization": f"Bearer {MOLTED_API_KEY}"},
        timeout=10,
    )
    resp.raise_for_status()
    result = resp.json()

    if result["status"] == "blocked":
        return {
            "status": "blocked",
            "blockReason": result["blockReason"],
            "requestId": result["requestId"],
            "message": f"Email blocked: {result['blockReason']}. Do not retry this send.",
        }

    return {
        "status": result["status"],
        "requestId": result["requestId"],
        "message": f"Email queued successfully. requestId={result['requestId']}",
    }

from google.adk.agents import Agent
from tools.send_email import send_email


outreach_agent = Agent(
    name="outreach_agent",
    model="gemini-2.0-flash",
    description="Sends governed outreach emails on behalf of the team.",
    instruction="""
    You are a customer outreach agent. When sending emails:
    - Always supply a dedupe_key using the format "{recipient}-{event}" to prevent duplicates.
    - If status is 'blocked', report the blockReason to the user and stop - do not retry.
    - 'duplicate_send': already sent recently - inform the user.
    - 'suppressed_recipient': contact has opted out - do not email them again.
    - 'rate_limit_exceeded': send rate reached - wait and inform the user.
    - 'risk_budget_exceeded': agent risk budget exhausted for this period.
    """,
    tools=[send_email],
)

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai.types import Content, Part
from agent import outreach_agent


async def main():
    session_service = InMemorySessionService()
    runner = Runner(
        agent=outreach_agent,
        app_name="outreach",
        session_service=session_service,
    )

    session = await session_service.create_session(
        app_name="outreach",
        user_id="user-1",
    )

    async for event in runner.run_async(
        user_id="user-1",
        session_id=session.id,
        new_message=Content(
            role="user",
            parts=[
                Part(text=(
                    "Send a trial welcome email to alice@example.com. "
                    "Subject: 'Welcome to the trial' "
                    "Body: 'Thanks for signing up - your 14-day trial starts now.' "
                    "Use dedupe key: alice-trial-welcome"
                ))
            ],
        ),
    ):
        if event.is_final_response() and event.content:
            for part in event.content.parts:
                if part.text:
                    print(part.text)


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

import { Agent, tool } from "@google/adk/agents";
import { z } from "zod";

const MOLTED_API_KEY = process.env.MOLTED_API_KEY!;
const MOLTED_TENANT_ID = process.env.MOLTED_TENANT_ID!;
const MOLTED_MAILBOX_ID = process.env.MOLTED_MAILBOX_ID!;

const sendEmail = tool({
  name: "send_email",
  description:
    "Send an email to a contact through the governed mailbox. " +
    "The email is evaluated against policy rules before delivery. " +
    "Blocked sends return a reason and should not be retried.",
  inputSchema: z.object({
    to: z.string().describe("Recipient email address"),
    subject: z.string().describe("Email subject line"),
    body: z.string().describe("Email body in plain text or HTML"),
    dedupeKey: z
      .string()
      .optional()
      .describe(
        "Unique key to prevent duplicate sends. Use format: {recipient}-{event}"
      ),
  }),
  async execute({ to, subject, body, dedupeKey }) {
    const resp = await fetch("https://api.molted.email/v1/agent/send", {
      method: "POST",
      headers: {
        Authorization: `Bearer ${MOLTED_API_KEY}`,
        "Content-Type": "application/json",
      },
      body: JSON.stringify({
        tenantId: MOLTED_TENANT_ID,
        mailboxId: MOLTED_MAILBOX_ID,
        recipientEmail: to,
        templateId: "_default",
        dedupeKey: dedupeKey ?? `${to}-${subject}`,
        agentId: "google-adk-agent",
        payload: { subject, html: body, text: body },
      }),
    });

    const result = await resp.json();

    if (result.status === "blocked") {
      return {
        status: "blocked",
        blockReason: result.blockReason,
        requestId: result.requestId,
        message: `Email blocked: ${result.blockReason}. Do not retry.`,
      };
    }

    return {
      status: result.status,
      requestId: result.requestId,
      message: `Email queued. requestId=${result.requestId}`,
    };
  },
});

export const outreachAgent = new Agent({
  name: "outreach_agent",
  model: "gemini-2.0-flash",
  description: "Sends governed outreach emails on behalf of the team.",
  instruction: `
    You are a customer outreach agent. When sending emails:
    - Always supply a dedupeKey using the format "{recipient}-{event}" to prevent duplicates.
    - If status is 'blocked', report the blockReason to the user and stop - do not retry.
    - 'duplicate_send': already sent recently.
    - 'suppressed_recipient': contact has opted out - do not email them again.
    - 'rate_limit_exceeded': send rate reached - wait and inform the user.
  `,
  tools: [sendEmail],
});

from google.adk.agents import Agent
from tools.send_email import send_email


# Email specialist - the only agent with mailbox access
email_agent = Agent(
    name="email_agent",
    model="gemini-2.0-flash",
    description="Sends and tracks emails. Only this agent has mailbox access.",
    instruction="""
    You send emails for the team. Always use a dedupe_key.
    Report policy blocks clearly - do not retry terminal blocks.
    """,
    tools=[send_email],
)

# Coordinator - routes tasks and delegates email sends
coordinator_agent = Agent(
    name="coordinator",
    model="gemini-2.0-flash",
    description="Coordinates outreach tasks and delegates email sends to the email agent.",
    instruction="""
    You coordinate outreach tasks. For any email sending, delegate to the email_agent.
    Do not attempt to send emails directly.
    """,
    sub_agents=[email_agent],
)

import os
import requests


MOLTED_API_KEY = os.environ["MOLTED_API_KEY"]


def check_email_status(request_id: str) -> dict:
    """Check the delivery status of a previously queued email.

    Args:
        request_id: The requestId returned when the email was sent.

    Returns:
        A dict with keys: status, provider, lastEvent.
    """
    resp = requests.get(
        f"https://api.molted.email/v1/agent/send/{request_id}/status",
        headers={"Authorization": f"Bearer {MOLTED_API_KEY}"},
        timeout=10,
    )
    resp.raise_for_status()
    data = resp.json()
    return {
        "status": data["status"],
        "provider": data.get("provider", "unknown"),
        "lastEvent": data.get("lastEvent", "none"),
    }

molted listen --pipe "python reply_handler.py" --events "inbound.*"

import sys
import json
import asyncio
import os
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai.types import Content, Part
from tools.send_email import send_email


reply_agent = Agent(
    name="reply_agent",
    model="gemini-2.0-flash",
    description="Handles inbound email replies.",
    instruction="You handle inbound replies. Respond helpfully and concisely.",
    tools=[send_email],
)

session_service = InMemorySessionService()
runner = Runner(
    agent=reply_agent,
    app_name="reply",
    session_service=session_service,
)


async def handle_event(event: dict) -> None:
    if event.get("event") != "inbound.classified":
        return

    data = event["data"]
    if data.get("intent") == "support" and data.get("suggestedAction") == "reply":
        session = await session_service.create_session(
            app_name="reply",
            user_id="inbound",
        )
        async for adk_event in runner.run_async(
            user_id="inbound",
            session_id=session.id,
            new_message=Content(
                role="user",
                parts=[Part(text=f"Reply to {data['fromEmail']} who wrote: '{data['subject']}'.")],
            ),
        ):
            if adk_event.is_final_response() and adk_event.content:
                for part in adk_event.content.parts:
                    if part.text:
                        print(f"Reply sent: {part.text}")


for line in sys.stdin:
    event_data = json.loads(line.strip())
    asyncio.run(handle_event(event_data))