Amazon SESで受信したメールをトリガーに、AWS Lambdaで本文の電話番号へAmazon Connectから自動発信してみた

外部メール
↓
Amazon SES Eメール受信
↓
Amazon S3 に raw email を保存
↓
S3 イベント通知
↓
Amazon SQS
↓
AWS Lambda
↓
Amazon Connect StartOutboundVoiceContact
↓
メール本文の電話番号宛に電話を発信

Please call +819012345678

Amazon SES Eメール受信
↓
受信ルール
↓
S3 に raw email 保存
↓
S3 イベント通知
↓
SQS 標準キュー
↓
Lambda
↓
Amazon Connect StartOutboundVoiceContact
↓
メール本文の電話番号宛に電話を発信

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowSESPutObject",
      "Effect": "Allow",
      "Principal": {
        "Service": "ses.amazonaws.com"
      },
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::my-ses-inbound-mail-bucket/incoming/*",
      "Condition": {
        "StringEquals": {
          "AWS:SourceAccount": "111111111111"
        }
      }
    }
  ]
}

{
  "Sid": "AllowS3SendMessage",
  "Effect": "Allow",
  "Principal": {
    "Service": "s3.amazonaws.com"
  },
  "Action": "SQS:SendMessage",
  "Resource": "arn:aws:sqs:ap-northeast-1:111111111111:connect-rate-limited-call-queue",
  "Condition": {
    "StringEquals": {
      "aws:SourceAccount": "111111111111"
    },
    "ArnLike": {
      "aws:SourceArn": "arn:aws:s3:::my-ses-inbound-mail-bucket"
    }
  }
}

Unable to validate the following destination configurations

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ReadInboundMailFromS3",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject"
      ],
      "Resource": "arn:aws:s3:::my-ses-inbound-mail-bucket/incoming/*"
    },
    {
      "Sid": "StartOutboundVoiceContact",
      "Effect": "Allow",
      "Action": [
        "connect:StartOutboundVoiceContact"
      ],
      "Resource": "arn:aws:connect:ap-northeast-1:111111111111:instance/<connect-instance-id>/contact/*"
    }
  ]
}

import json
import os
import re
import time
import hashlib
import urllib.parse
from email import policy
from email.parser import BytesParser

import boto3
from botocore.config import Config

s3 = boto3.client("s3")

connect = boto3.client(
    "connect",
    config=Config(
        retries={
            "total_max_attempts": 1,
            "mode": "standard"
        }
    )
)

CONNECT_INSTANCE_ID = os.environ["CONNECT_INSTANCE_ID"]
CONTACT_FLOW_ID = os.environ["CONTACT_FLOW_ID"]
SOURCE_PHONE_NUMBER = os.environ["SOURCE_PHONE_NUMBER"]
DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true"

PHONE_RE = re.compile(r"\+[1-9]\d{1,14}\b")

def lambda_handler(event, context):
    print("event:", json.dumps(event, ensure_ascii=False))

    for sqs_record in event.get("Records", []):
        body = json.loads(sqs_record["body"])

        if body.get("Event") == "s3:TestEvent":
            print("Received s3:TestEvent. Ignore.")
            continue

        for s3_record in body.get("Records", []):
            bucket = s3_record["s3"]["bucket"]["name"]
            key = urllib.parse.unquote_plus(s3_record["s3"]["object"]["key"])

            print(f"Processing s3://{bucket}/{key}")

            raw_email = get_raw_email(bucket, key)
            subject, text = parse_email(raw_email)

            destination_phone_number = extract_phone_number(subject + "\n" + text)

            if not destination_phone_number:
                print(f"No E.164 phone number found in s3://{bucket}/{key}. Skip.")
                continue

            client_token = make_client_token(bucket, key)

            print({
                "bucket": bucket,
                "key": key,
                "subject": subject,
                "destinationPhoneNumber": destination_phone_number,
                "clientToken": client_token,
                "dryRun": DRY_RUN
            })

            if DRY_RUN:
                print("DRY_RUN=true, skip StartOutboundVoiceContact.")
            else:
                start_outbound_voice_contact(
                    destination_phone_number=destination_phone_number,
                    client_token=client_token,
                    bucket=bucket,
                    key=key,
                    subject=subject
                )

                time.sleep(1)

    return {
        "statusCode": 200
    }

def get_raw_email(bucket, key):
    obj = s3.get_object(Bucket=bucket, Key=key)
    return obj["Body"].read()

def parse_email(raw_email_bytes):
    msg = BytesParser(policy=policy.default).parsebytes(raw_email_bytes)

    subject = msg.get("subject", "")

    texts = []

    if msg.is_multipart():
        for part in msg.walk():
            content_type = part.get_content_type()
            content_disposition = part.get_content_disposition()

            if content_disposition == "attachment":
                continue

            if content_type == "text/plain":
                try:
                    texts.append(part.get_content())
                except Exception as e:
                    print(f"Failed to read text/plain part: {e}")
    else:
        try:
            if msg.get_content_type() == "text/plain":
                texts.append(msg.get_content())
            else:
                texts.append(str(msg.get_content()))
        except Exception as e:
            print(f"Failed to read message content: {e}")

    return subject, "\n".join(texts)

def extract_phone_number(text):
    match = PHONE_RE.search(text)
    if not match:
        return None
    return match.group(0)

def make_client_token(bucket, key):
    raw = f"{bucket}/{key}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

def start_outbound_voice_contact(destination_phone_number, client_token, bucket, key, subject):
    params = {
        "InstanceId": CONNECT_INSTANCE_ID,
        "ContactFlowId": CONTACT_FLOW_ID,
        "DestinationPhoneNumber": destination_phone_number,
        "SourcePhoneNumber": SOURCE_PHONE_NUMBER,
        "ClientToken": client_token,
        "Attributes": {
            "ClientToken": client_token,
            "S3Bucket": bucket,
            "S3Key": key[:1024],
            "MailSubject": subject[:1024]
        }
    }

    response = connect.start_outbound_voice_contact(**params)
    print("StartOutboundVoiceContact response:", response)
    return response

PHONE_RE = re.compile(r"\+[1-9]\d{1,14}\b")

+819012345678

090-1234-5678
09012345678
+81 90 1234 5678

bucket + "/" + objectKey

def make_client_token(bucket, key):
    raw = f"{bucket}/{key}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

connect = boto3.client(
    "connect",
    config=Config(
        retries={
            "total_max_attempts": 1,
            "mode": "standard"
        }
    )
)

Please call +819012345678

Processing s3://my-ses-inbound-mail-bucket/incoming/4cklatksvuq0qc90g1k73lufosj88q5vueag8eg1

{
  'bucket': 'my-ses-inbound-mail-bucket',
  'key': 'incoming/4cklatksvuq0qc90g1k73lufosj88q5vueag8eg1',
  'subject': 'テストです。',
  'destinationPhoneNumber': '+819012345678',
  'clientToken': '9ec31a9179cce2d6ddfafc1b228ff4721a1cbe41f6f6f050d4a7a18ec4f6abc8',
  'dryRun': True
}

DRY_RUN=true, skip StartOutboundVoiceContact.

- DRY_RUN=true
+ DRY_RUN=false

メール送信
↓
SES で受信
↓
S3 に raw email 保存
↓
S3 イベント通知で SQS に通知
↓
Lambda 起動
↓
メール本文から電話番号を抽出
↓
StartOutboundVoiceContact API 呼び出し
↓
電話発信