Enterprise Email Agents on Google Vertex AI

import vertexai
from vertexai.generative_models import (
    GenerativeModel, Tool, FunctionDeclaration
)
import requests

vertexai.init(project="your-gcp-project", location="us-central1")
MULTIMAIL_API = "https://api.multimail.dev/v1"
MM_HEADERS = {"Authorization": "Bearer mm_live_your_api_key"}

send_email_fn = FunctionDeclaration(
    name="send_email",
    description="Send an email through MultiMail. In gated_send mode, queues for human approval.",
    parameters={
        "type": "object",
        "properties": {
            "mailbox_id": {"type": "string", "description": "Mailbox to send from"},
            "to": {"type": "string", "description": "Recipient email"},
            "subject": {"type": "string", "description": "Subject line"},
            "body": {"type": "string", "description": "Email body"}
        },
        "required": ["mailbox_id", "to", "subject", "body"]
    }
)

check_inbox_fn = FunctionDeclaration(
    name="check_inbox",
    description="Check inbox for recent messages.",
    parameters={
        "type": "object",
        "properties": {
            "mailbox_id": {"type": "string", "description": "Mailbox to check"},
            "limit": {"type": "integer", "description": "Max messages"}
        },
        "required": ["mailbox_id"]
    }
)

email_tool = Tool(function_declarations=[send_email_fn, check_inbox_fn])

from vertexai.generative_models import Part, Content

def execute_function(name, args):
    if name == "send_email":
        resp = requests.post(f"{MULTIMAIL_API}/send", headers=MM_HEADERS, json=dict(args))
    elif name == "check_inbox":
        resp = requests.get(
            f"{MULTIMAIL_API}/mailboxes/{args[&"cm">#039;mailbox_id']}/inbox",
            headers=MM_HEADERS, params={"limit": args.get("limit", 10)}
        )
    elif name == "reply_email":
        resp = requests.post(f"{MULTIMAIL_API}/reply", headers=MM_HEADERS, json=dict(args))
    else:
        return {"error": f"Unknown function: {name}"}
    return resp.json()

model = GenerativeModel(
    model_name="gemini-2.0-flash",
    tools=[email_tool],
    system_instruction="You are an email assistant. Emails use gated_send mode "
    "and queue for human approval before delivery."
)

chat = model.start_chat()
response = chat.send_message("Check my inbox and summarize new messages")

while response.candidates[0].content.parts[0].function_call.name:
    fc = response.candidates[0].content.parts[0].function_call
    result = execute_function(fc.name, fc.args)
    response = chat.send_message(
        Part.from_function_response(name=fc.name, response={"result": result})
    )

print(response.text)

import functions_framework
from google.cloud import secretmanager
import requests
import json

def get_api_key():
    client = secretmanager.SecretManagerServiceClient()
    name = "projects/your-project/secrets/multimail-api-key/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return response.payload.data.decode("UTF-8")

@functions_framework.http
def handle_email_tool(request):
    """Cloud Function for Vertex AI Agent Builder email tool."""
    data = request.get_json()
    action = data.get("action")
    params = data.get("parameters", {})

    api_key = get_api_key()
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    api = "https://api.multimail.dev/v1"

    if action == "send_email":
        resp = requests.post(f"{api}/send", headers=headers, json=params)
    elif action == "check_inbox":
        mailbox_id = params.pop("mailbox_id")
        resp = requests.get(f"{api}/mailboxes/{mailbox_id}/inbox",
                           headers=headers, params=params)
    elif action == "reply_email":
        resp = requests.post(f"{api}/reply", headers=headers, json=params)
    else:
        return json.dumps({"error": "Unknown action"}), 400

    return json.dumps(resp.json()), 200