เริ่มต้นใช้งาน Mailtarget + Google Vertex AI

# Install gcloud CLI (if not already installed) 
# https://cloud.google.com/sdk/docs/install  

# Authenticate 
gcloud auth login 
gcloud auth application-default login  

# Set your project 
gcloud config set project YOUR_PROJECT_ID  

# Enable Vertex AI API gcloud services enable aiplatform.googleapis.com

export MAILTARGET_API_KEY=your_api_key_here

pip install google-cloud-aiplatform requests

import requests
import os

def send_email(
    to_email: str,
    to_name: str,
    subject: str,
    body_html: str,
    body_text: str = ""
) -> dict:
    """Send an email via Mailtarget API.

    Args:
        to_email: Recipient email address.
        to_name: Recipient name.
        subject: Email subject line.
        body_html: HTML content of the email.
        body_text: Plain text fallback (optional).

    Returns:
        dict with transmissionId on success, or error details.
    """
    response = requests.post(
        "https://transmission.mailtarget.co/v1/layang/transmissions",
        headers={
            "Authorization": f"Bearer {os.environ['MAILTARGET_API_KEY']}",
            "Content-Type": "application/json"
        },
        json={
            "to": [{"email": to_email, "name": to_name}],
            "from": {
                "email": "[email protected]",
                "name": "Your Company"
            },
            "subject": subject,
            "bodyHtml": body_html,
            "bodyText": body_text,
            "optionsAttributes": {
                "openTracking": True,
                "clickTracking": True,
                "transactional": True
            }
        }
    )
    return response.json()

from google.adk.agents import Agent

agent = Agent(
    name="email_agent",
    model="gemini-2.0-flash",
    description="An agent that sends emails via Mailtarget.",
    instruction="""You are an email assistant. When asked to send 
    an email, use the send_email tool. Write professional HTML 
    emails with clean formatting. Always confirm the transmissionId 
    after sending.""",
    tools=[send_email],
)

# Run the agent
response = agent.run(
    "Send a welcome email to [email protected] (John Smith) "
    "with a friendly onboarding message."
)
print(response)

from langchain_google_vertexai import ChatVertexAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate

@tool
def send_email(to_email: str, to_name: str, subject: str, 
               body_html: str, body_text: str = "") -> dict:
    """Send an email via Mailtarget API."""
    import requests, os
    response = requests.post(
        "https://transmission.mailtarget.co/v1/layang/transmissions",
        headers={
            "Authorization": f"Bearer {os.environ['MAILTARGET_API_KEY']}",
            "Content-Type": "application/json"
        },
        json={
            "to": [{"email": to_email, "name": to_name}],
            "from": {"email": "[email protected]", 
                     "name": "Your Company"},
            "subject": subject,
            "bodyHtml": body_html,
            "bodyText": body_text,
            "optionsAttributes": {"openTracking": True, 
                                   "clickTracking": True}
        }
    )
    return response.json()

llm = ChatVertexAI(model="gemini-2.0-flash")

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an email assistant. Use the send_email "
               "tool when asked to send emails."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, [send_email], prompt)
executor = AgentExecutor(agent=agent, tools=[send_email])

result = executor.invoke({
    "input": "Send a welcome email to [email protected] (John Smith) "
             "with a friendly onboarding message."
})
print(result["output"])

import vertexai
from vertexai import agent_engines

vertexai.init(project="YOUR_PROJECT_ID", location="us-central1")

# Deploy the agent
remote_agent = agent_engines.create(
    agent_engine=agent,
    requirements=["requests"],
    display_name="Mailtarget Email Agent",
)

# Query the deployed agent
session = remote_agent.create_session(user_id="user-1")
response = remote_agent.stream_query(
    session=session,
    message="Send a test email to [email protected]"
)
for chunk in response:
    print(chunk)