1. 目的

AWSアカウントの管理を簡単に
AWSアカウントの利用が増えると、毎日の利用料金をチェックするのが大変になってきます。
手動で確認するのは時間がかかるだけでなく、ミスも起こりやすいです。
そこで、自動化されたシステムを使って利用料金を簡単に取得・管理し、作業の手間を減らしたいと考えています。
チーム全体でコストを共有
利用料金の情報をチーム全員と共有することで、無駄なリソースの使用を防ぎ、コスト意識を高めることができます。
Teamsに自動で投稿する仕組みを作ることで、全員が最新のコスト状況を確認できるようになります。

2. システム概要

システムは以下のサービスで構成しています。

Amazon EventBridge (Scheduler)
定期的なスケジュールに基づき、Lambda関数を自動実行します。
AWS Lambda
利用料金データの取得と処理を行い、Teamsへ情報を送信します。
AWS Identity and Access Management (IAM)
各サービス間のアクセス権限を適切に管理し、セキュリティを確保します。
AWS Key Management Service (KMS)
Lambda関数で使用するシークレット情報（TeamsのWebhook URL）の暗号化に使用します。
Amazon CloudWatch (Logs)
Lambda関数の実行ログやエラーログをCloudWatch Logsに送信し、一元管理します。
Power Automate (WorkFlows)
受信したPOSTリクエストを基に、チームメンバーがアクセスするチャネルに、利用料金情報を表示します。

3. 通知例

完成したシステムの通知例です。

4. 作成方法

4-1. WorkFlows作成

投稿するチャネルのWebhook URLを取得します。
手順は以下のページを参考にさせていただきました。

対象のチャネルをTeamsで開き、その他のオプションから「ワークフロー」をクリックします。

ワークフローの選択画面で「Webhook 要求を受信するとチャネルに投稿する」を選択します。

作成するワークフローの名前を入力します。今回は「AWS利用料金投稿」としました。
名前の入力が完了したら「次へ」をクリックします。

投稿先のチームとチャネルを選択します。
手順通り進めていれば、初期値に対象のチームとチャネルが選択されているはずです。投稿先を変更したい場合は、ここで変更してください。

プライベートチャネルには投稿できないので、注意してください。

投稿先の選択が完了したら「ワークフローを追加する」をクリックします。

ワークフローが追加され、Webhook URLが払い出されました。
Webhook URLはこの後の作業で使用するため、コピーしておきます。
コピーが完了したら「完了」を選択してダイアログを閉じます。

4.2. IAMロールの作成

Lambda用のIAMロールを作成します。

AWSコンソールで「IAM」に移動し、「ロールの作成」をクリックします。

Lambdaをサービスとして選択し、「次へ」をクリックします。

以下の許可ポリシーを選択し、「次へ」をクリックします。

AWSAccountUsageReportAccess
AWSBillingReadOnlyAccess
AWSLambdaBasicExecutionRole

ロール名と説明（任意）を入力します。必要であれば、タグも追加してください。
今回のロール名は「AWSCostNotifications-LambdaExecutionRole」としました。

ロール名と説明の入力、タグの追加が完了したら、ここまでの設定項目を確認し、「ロールを作成」をクリックします。

ロールが作成されたことを確認します。ロールのARNはこの後の作業で使用するため、コピーしておきます。

4.3. KMSキーの作成・設定

Lambdaの環境変数を暗号化するためのKMSキーを作成します。

AWSコンソールで「KMS」に移動し、「カスタマー管理型のキー」を選択、「キーの作成」をクリックします。

キーのタイプ：「対称」
キーの使用法：「暗号化及び複合化」
キーマテリアルオリジン：「KMS - 推奨」
リージョンごと：「単一リージョンキー」を選択し、「次へ」をクリックします。

エイリアスと説明（任意）を入力します。必要であれば、タグも追加してください。
今回のエイリアスは「AWSCostNotifications-LambdaKey」としました。
エイリアスと説明の入力、タグの追加が完了したら、「次へ」をクリックします。

キー管理者を設定します。今回は自身のIAMユーザを追加しました。キーの削除にもチェックを入れておきます。
設定が完了したら、「次へ」をクリックします。

その他の設定は変更しないため、確認画面まで進みます。
ここまでの設定項目を確認し、「完了」をクリックします。

KMSキーが作成されたことを確認します。

作成したKMSキーの画面を開き、ARNをコピーします。

キーポリシーの編集画面を開き、以下のポリシーを追加します。

{
      "Sid": "AWSCostNotifications-LambdaExecutionRole",
      "Effect": "Allow",
      "Principal": {
        "AWS": "<手順4.2でコピーしたIAMロールのARN>",
        "Service": "lambda.amazonaws.com"
      },
      "Action": [
        "kms:Decrypt",
        "kms:GenerateDataKey"
      ],
      "Resource": "<先程コピーしたKMSキーのARN>"
}

4.4. IAMポリシーの作成

KMSキーアクセス用のIAMポリシーを作成し、手順4.2で作成したLambda用のIAMロールにアタッチします。

AWSコンソールで「IAM」に移動し、「ポリシーの作成」をクリックします。

以下のポリシーを入力し、「次へ」をクリックします。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kms:Decrypt",
                "kms:GenerateDataKey"
            ],
            "Resource": "<手順4.3でコピーしたKSMキーのARN>"
        }
    ]
}

ポリシー名と説明（任意）を入力します。必要であれば、タグも追加してください。
今回のポリシー名は「AWSCostNotifications-KMSAccessPolicy」としました。

ポリシー名と説明の入力、タグの追加が完了したら、ここまでの設定項目を確認し、「ポリシーを作成」をクリックします。

ポリシーが作成されたことを確認します。
4.2で作成したLambda用のIAMロールにアタッチします。

4.5. Lambda関数の作成

AWSコンソールで「Lambda」に移動し、「関数を作成」をクリックします。

「一から作成」を選択します。
関数名を入力します。今回の関数名は「AWSCostNotifications-CostNotifierFunction」としました。
ランタイムは「Python 3.13」、アーキテクチャは「x86_64」を選択します。

実行ロールは「既存のロールを使用する」を選択し、手順4.2で作成したIAMロールを選択します。

「AWS KMS カスタマーマネージドキーによる暗号化を有効にする」にチェックを入れ、手順4.3で作成したKMSキーを選択します。
必要であれば、「タグを有効化」にチェックを入れ、タグも追加してください。

関数が作成されたことを確認します。
このまま、追加の設定をしていきます。

コードソースに以下のコードを貼り付け、デプロイします。
コードは以下のページを参考にさせていただきました。

Lambdaコード

import os
import boto3
import json
import requests
import logging
import botocore.exceptions
import time
import random
from datetime import datetime, timedelta, date
from base64 import b64decode

logger = logging.getLogger()
logger.setLevel(logging.INFO)

formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')

if not logger.handlers:
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)

ce_client = boto3.client('ce', region_name='ap-northeast-1')
kms_client = boto3.client('kms')

def main(event, context) -> None:
    try:
        logger.info("Processing AWS billing started.")
        
        decrypted_url = decrypt(os.environ['TEAMS_WEBHOOK_URL'], context.function_name)
        
        total_cost, svc_costs = get_costs(ce_client)
        
        title, detail = create_message(total_cost, svc_costs)
        post_to_teams(title, detail, svc_costs, decrypted_url)
        logger.info("Processing AWS billing completed successfully.")
        
    except Exception:
        logger.exception("An error occurred during processing.")

def decrypt(enc_url: str, func_name: str) -> str:
    try:
        logger.info("Starting URL decryption.")
        
        decrypted = kms_client.decrypt(
            CiphertextBlob=b64decode(enc_url),
            EncryptionContext={'LambdaFunctionName': func_name}
        )['Plaintext'].decode('utf-8')

        logger.info("URL decryption completed.")
        return decrypted

    except Exception:
        logger.exception("Error occurred during URL decryption.")
        raise

def retry_with_backoff(func, retries=5, backoff_in_seconds=1, factor=2):
    current_try = 0
    while current_try <= retries:
        try:
            return func()
        except botocore.exceptions.ClientError as error:
            error_code = error.response['Error']['Code']
            if error_code in ['ThrottlingException', 'LimitExceededException']:
                sleep = backoff_in_seconds * (factor ** current_try) + random.uniform(0, 1)
                logger.warning(f"Rate limit exceeded. Retrying in {sleep:.2f} seconds...")
                time.sleep(sleep)
                current_try += 1
            else:
                raise
    logger.error("Max retries exceeded.")
    raise Exception("Max retries exceeded.")

def get_costs(client: boto3.client) -> tuple[dict, list[dict]]:
    start, end, is_prev_month = get_range()
    
    def fetch_cost_data():
        logger.info(f"Fetching cost data: {start} - {end}")
        
        return client.get_cost_and_usage(
            TimePeriod={'Start': start, 'End': end},
            Granularity='MONTHLY',
            Metrics=['UnblendedCost'],
            GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
        )
    
    try:
        response = retry_with_backoff(fetch_cost_data)
        logger.info("Cost data fetched successfully.")

        svc_costs = [
            {'service': grp['Keys'][0], 'billing': grp['Metrics']['UnblendedCost']['Amount']}
            for grp in response['ResultsByTime'][0].get('Groups', [])
            if float(grp['Metrics']['UnblendedCost']['Amount']) > 0.0
        ]

        total_amount = sum(float(grp['Metrics']['UnblendedCost']['Amount']) for grp in response['ResultsByTime'][0].get('Groups', []))
        
        total_cost = {
            'start': response['ResultsByTime'][0]['TimePeriod']['Start'],
            'end': response['ResultsByTime'][0]['TimePeriod']['End'],
            'billing': f"{total_amount:.2f}",
        }
        
        return total_cost, svc_costs

    except Exception:
        logger.exception("Error occurred while fetching cost data.")
        raise

def create_message(total_cost: dict, svc_costs: list[dict]) -> tuple[str, str]:
    start_str, end_str, is_prev_month = get_range()
    start = datetime.strptime(start_str, "%Y-%m-%d").strftime("%m/%d")
 
    if is_prev_month:
        end = datetime.strptime(end_str, "%Y-%m-%d").strftime("%m/%d")
    else:
        end = (datetime.strptime(end_str, "%Y-%m-%d") - timedelta(days=1)).strftime("%m/%d")
 
    title = f"{start}～{end}の請求額は、{float(total_cost['billing']):.2f} USDです。"
    detail = '\n'.join(f'　・{s["service"]}: {float(s["billing"]):.2f} USD' for s in svc_costs)
 
    return title, detail

def post_to_teams(title: str, detail: str, services: list[dict], url: str) -> None:
    try:
        logger.info("Preparing Teams message.")
        
        svc_details = []
        for item in services:
            svc_name = item['service']
            billing = round(float(item['billing']), 2)
            if billing == 0.0:
                continue
            svc_details.append({'name': f'{billing:.2f} USD', 'value': svc_name})

        sorted_svc_details = sorted(svc_details, key=lambda x: float(x['name'].rstrip(' USD')), reverse=True)

        card_content = create_card(title, sorted_svc_details)
        payload = {
            "type": "message",
            "attachments": [
                {
                    "contentType": "application/vnd.microsoft.card.adaptive",
                    "content": card_content
                }
            ]
        }

        response = requests.post(url, headers={"Content-Type": "application/json"}, data=json.dumps(payload))
        response.raise_for_status()

        logger.info(f"Message posted to Teams successfully: {response.status_code}")

    except requests.exceptions.RequestException:
        logger.exception("Error occurred while posting to Teams.")
        raise

def create_card(title: str, facts: list[dict[str, str]]) -> dict:
    return {
        "type": "AdaptiveCard",
        "version": "1.4",
        "body": [
            {
                "type": "Container",
                "items": [
                    {
                        "type": "ColumnSet",
                        "columns": [
                            {
                                "type": "Column",
                                "items": [
                                    {
                                        "type": "TextBlock",
                                        "size": "Medium",
                                        "weight": "Bolder",
                                        "text": title
                                    },
                                    {
                                        "type": "TextBlock",
                                        "size": "Small",
                                        "text": "サービス別利用金額(金額降順)",
                                        "isSubtle": True
                                    },
                                    {
                                        "type": "ColumnSet",
                                        "columns": [
                                            {
                                                "type": "Column",
                                                "width": "auto",
                                                "items": [
                                                    {
                                                        "type": "TextBlock",
                                                        "size": "Small",
                                                        "text": fact['value'],
                                                        "wrap": True
                                                    }
                                                    for fact in facts
                                                ]
                                            },
                                            {
                                                "type": "Column",
                                                "width": "auto",
                                                "items": [
                                                    {
                                                        "type": "TextBlock",
                                                        "size": "Small",
                                                        "weight": "Bolder",
                                                        "text": fact['name'],
                                                        "wrap": True
                                                    }
                                                    for fact in facts
                                                ]
                                            }
                                        ]
                                    }
                                ]
                            }
                        ]
                    }
                ]
            }
        ],
        "$schema": "https://adaptivecards.io/schemas/adaptive-card.json"
    }

def get_range() -> tuple[str, str, bool]:
    today = date.today()
    start = today.replace(day=1).isoformat()
    end = today.isoformat()
    is_prev_month = False

    if today.day == 1:
        last_day = today - timedelta(days=1)
        start = last_day.replace(day=1).isoformat()
        end = last_day.isoformat()
        is_prev_month = True

    return start, end, is_prev_month

今回のコードでは、requestsモジュールが必要となります。
requestsモジュールのLambda Layerを作成し、Lambda関数に追加します。

手順は以下のページを参考にさせていただきました。

環境変数を設定します。

キーに「TEAMS_WEBHOOK_URL」、値に「手順4.1で作成したWebhook URL」を設定します。
環境変数はKMSキーで暗号化してください。

その他の暗号化の設定の説明は割愛します。

4.6. EventBridgeの設定

AWSコンソールで「EventBridge」に移動し、「ルールを作成」をクリックします。

名前と説明（任意）を入力します。
今回の名前は「AWSCostNotifications-ScheduleRule」としました。
イベントバスは「default」、ルールタイプは「イベントパターンを持つルール」を選択し、「続行してルールを作成する」をクリックします。

スケジュールを定義します。
今回は毎日9時と17時（日本時間）に通知してほしいため、「0 0,8 ? * * *」と定義しました。
定義が完了したら、「次へ」をクリックします。

ターゲットとして、手順4.5で作成したLambda関数を選択します。
選択が完了したら、「次へ」をクリックします。

必要であれば、タグも追加してください。
ここまでの設定項目を確認し、「ルールを作成」をクリックします。

5. テストと検証

システム全体が正しく動作することを確認するために、以下のテストを実施します。

5.1. Lambda関数の動作確認

手動テスト
Lambdaコンソールから手動で関数を実行し、Teamsにメッセージが投稿されるか確認します。
ログの確認
CloudWatch LogsでLambda関数のログを確認し、エラーがないことを確認します。

5.2. EventBridgeの動作確認

スケジュールテスト
EventBridgeのルールを手動でトリガーし、Lambda関数が正しく起動するか確認します。
定期実行の確認
実際のスケジュールに基づいて定期的に実行されることを確認します。

5.3. 通知内容の検証

メッセージの内容確認
Teamsに投稿されるメッセージが正確な利用料金データを反映しているか確認します。
フォーマットの確認
メッセージのフォーマットが読みやすく、必要な情報が含まれているか確認します。

6. まとめ

本記事では、EventBridge、Lambda、Power Automateを使用して、AWSの利用料金をTeamsに自動投稿するシステムの構築方法について解説しました。
本記事が少しでもお役に立てれば幸いです。