NRIネットコム社員が様々な視点で、日々の気づきやナレッジを発信するメディアです

注目のタグ

    手動からの解放!!Strands Agentsで実現する総合テスト自動化

    はじめに

    こんにちは。インフラエンジニア二年目の井手亮太です。最近、私は「総合テスト」という業務を経験しました。
    総合テストとは、システムの振る舞いや能力が、要件・仕様通りであるか確認することで、主に調査フェーズとエビデンス取得フェーズで構成されます。

    総合テスト概要

    具体的に何をしたのか

    私はこれまで、手動とAWS CLIの2つの方法で総合テストに取り組んできました。 どちらにも課題があり、正直なところかなり苦労しました。 

    実際に取り組んだテスト手法

    • 手動の場合の課題

      • 非常に時間がかかる
      • 人的ミスが発生しやすい
    • CLI の場合の課題

      • シェルの知識に乏しかったのでスクリプト実装に時間がかかる
      • シェルスクリプトで作成したツールが一時的な用途にとどまり、汎用性に欠ける

    これらの課題を解決するためにたどり着いたのが、 Strands Agentsという強力な武器。

    本ブログでは、このStrands Agentsを駆使して、総合テストの自動化に挑んだ実践記をお届けします。

    Strands Agents とは

    AWSが2025年5月にリリースしたStrands Agentsは、Pythonベースで柔軟かつ迅速にAIエージェントを構築できるオープンソースライブラリです。 以下のようにたったの三行でエージェントを構築することができます。

    from strands import Agent
    
    agent = Agent()
    agent("エージェンティックAIについて教えてください。")
    

    github.com

    特徴

    • 多様な基盤モデルに対応 : Amazon Bedrock、Anthropic、OpenAIなど多様なモデルを使用可能

    • 多様なツール : 豊富な組み込みツールが備わっており、MCPにも対応。また、@tool デコレーターを使うことで、独自のツールを開発・拡張することも可能。

    • 高度なエージェント機能 : マルチエージェントや自己改善型エージェント、リアルタイムストリーミング機能を提供

    Strands Agents については、弊社の志水さんが詳しいブログを出していますので、そちらをご覧ください。

    tech.nri-net.com

    テスト対象システムとテストケース

    今回は私が個人開発している、AWS最新ニュース通知アプリをテスト対象とすることにします。

    AWS NEWS通知アプリアーキテクチャ
    洗い出したテストケースの概要は以下の通りです。
    テストケース

    テストケースの詳細は以下をご覧ください。

    テストケースの詳細はこちら

    大分類 テスト手法 期待される結果 テスト結果
    定期実行 メトリクスを分析
    NameSpace: AWS/EVENTS
    Metrics_NAME:Invocations
    EventRule:r-ide-get-aws-news
    期間:直近1日
    period:60
    statistic:Average
    10時(JST)にInvocationsが1以上
    アプリログ CloudWatchのログのインサイトを使用する。
    対象ロググループ: /aws/lambda/r-ide-get-aws-news
    期間:現在時刻から遡って直近1日間(UTCベース)
    クエリ内容
    fieldとして@timestampと@messageを出力する。
    ログが表示されること
    アプリログ CloudWatchのログのインサイトを使用する。
    対象ロググループ: /aws/lambda/r-ide-get-aws-news
    期間:現在時刻から遡って直近1日間(UTCベース)
    クエリ内容
    fieldとして@timestampと@messageを出力。filter levelでERRORという文字列を検索
    ログが表示されないこと
    メトリクス メトリクスを分析
    NameSpace:AWS/Lambda
    FunctionName:r-ide-get-aws-news
    Metrics_name:Invocations
    期間:直近1日間
    period:60
    statistic:Average
    10時(JST)にInvocationsが1以上
    メトリクス メトリクスを分析
    NameSpace:AWS/Lambda
    FunctionName:r-ide-get-aws-news
    Metrics_name:Errors
    期間:直近1日間
    period:60
    statistic:Average
    10時(JST)にErrorsが0
    Slack通知 Slackの通知内容を確認
    チャンネル名:times-r-ide
    チャンネルID:xxxxxxxx
    期間:直近1日間
    10:00(JST)にニュースが通知されている。

    テストツールのアーキテクチャ

    テスト自動化ツールのアーキテクチャを以下に示します。

    テストケース01~03のアーキテクチャ

    テストケース04のアーキテクチャ

    今回のテスト自動化は、Strands Agents に標準搭載されているツール群と、Lambda MCP Server を組み合わせることで実現していきます。次章では、具体的な実装方法について詳しく解説していきます。

    前提条件

    ※ AWS CLI をインストールしたうえで認証情報の設定が必要です。
    ※デフォルトではAmazon BedrockのClaude 3.7 Sonnet(us-west-2) を利用します。 そのため、事前にBedrockモデルを有効化しておいてください
    ※ Python 3.10以上、及びuvはインストール済みとします。
    ※ Ubuntu 24.04.1 LTS を使用しています。

    事前準備

    Strands Agents のインストール、及び Slack の認証情報を設定します。

    事前準備詳細

    Strands Agents のインストール

    まずは、Strands Agents をインストールしていきます。既存のシェル環境のパッケージ競合が起きないために、仮想環境を構築しています。

    # 仮想環境の作成
    uv venv my-env
    source my-env/bin/activate # 仮想環境の有効化

    仮想環境を有効化した後は、Strands Agents をインストールします。

    uv pip install strands-agents strands-agents-tools strands-agents-builder

    Slackの準備

    今回はAIがSlackにアクセスするための認証トークンを発行する必要があります。
    まず、Slack API にアクセスし、「Create New APP」を押下します。

    api.slack.com

    「From Scratch」を選択し、アプリの名前と、インストールするワークスペースを選ぶ。

    OAuth & Permissionsを押下する。

    Scope⇒Bot Token Scopesで「ADD an OAuth Scope」でスコープを追加する。以下の権限を追加。

    • chat:write
    • reactions:write
    • channels:history
    • app_mentions:read
    • channels:read
    • reactions:read
    • groups:read
    • im:read
    • mpim:read

    ページ上部の緑色のボタン「Install to ワークスペース名」を押下する。

    Bot User Auth Token が発行されるので控えておく。

    調査したい Slack チャンネルに移動する
    チャンネルの詳細を開き、「インテグレーション」タブを選択。
    「アプリを追加する」で、先ほど作成したアプリを選択する。

    次に、SLACK_APP_TOKENを発行していく。Slack API の Socket Mode を選択。

    Enable Socket Mode を有効化して、任意のトークン名を入力する。トークンが発行されるので控えておく。
    a

    最後にターミナル上で環境変数を設定する。

    export SLACK_BOT_TOKEN=発行したBot User Auth Token
    export SLACK_APP_TOKEN=Socket Mode で発行したToken

    いざ実装

    以下に、今回取り上げるソースコードの全体を掲載します。

    ソースコード全体

    from strands import Agent
    from strands.tools.mcp import MCPClient
    from mcp import stdio_client, StdioServerParameters
    from strands_tools import current_time, file_write,file_read,editor,current_time,shell,slack
    
    # システムプロンプト
    SYSTEM_PROMPT ="""
    🎯 あなたは「AWSリソースの総合テスト」を実施するAIエージェントです。
    テストケースファイルを読み取り、あらゆるツールを用いてテストを実行してください。
    テスト手順は以下の通りです。
    
    ✅ テスト手順:
    - 指定されたファイルを読み取り、テスト手法を判断してください。
    ファイルには、以下のような必要な情報が含まれています:
        - メトリクスグラフを取得するためのパラメータ(例:namespace、metrics_name)
        - Logs Insights を実行するためのパラメータ(例:ロググループ名、クエリ)
        - slackのチャンネル名
    
    - 取得した情報をもとに、テストを実行してください。注意事項は以下の通りです:
    <共通事項>
        - テストを実行すると同時にエビデンス取得を行う必要があります。取得したエビデンスは
        「エビデンス」フォルダに格納してください。
        - シェルを使用する場合は、aws cli を実行してください。
        - エビデンスには、検索結果だけではなく、検索結果の根拠として、検索に使った具体的なaws cliのコマンドを表示してください。
        - 得られた結果と「期待される結果」を比較し、「テストケース.md」にいずれかを記録してください:
        - 条件を満たしていれば:「テスト結果」列に"OK"を記入する
        - 条件を満たしていなければ:「テスト結果」列に"NG"を記入する
    
    <テスト手法が「メトリクスの分析」の場合>
        - テスト手法が「メトリクスの分析」の場合は、以下二つのエビデンスを保存してください。
            - aws cliを用いて、そのメトリクスが何時に発生したかを記載する。また、実行したaws cliのコマンドを記載する。
            - r-ide-get-metricsgraphを実行しS3バケットからCLIを使って「エビデンス」フォルダーにメトリクスグラフをダウンロードする
    
    <テスト手法が「Slack通知の確認」の場合>
        - エビデンスは該当スレッドのURLと通知の内容をテキストファイルにまとめるようにしてください
    
    <注意点>
     AWS Lambda関数(r-ide-get-metricsgraph)を実行する際に、テストケースファイル内のパラメータとLambda関数の入力パラメータが一致している必要があります。
     テストケースのパラメータをLambda関数の期待する形式に変換・調整するようにしてください。なお、Lambda関数は以下のようなパラメータを期待しています。
     {
      "metrics_name": "メトリクス名",
      "namespace": "名前空間",
      "statistic": "Average",
      "period": 60,
      "hours": 24,
      "bucket_name": "S3バケット名",
      "dimensions": [
        {
          "Name": "ディメンジョンの名前",
          "Value": "ディメンジョンの値"
        }
      ]
    }
    """
    # 組み込みツールをリスト化する
    tools = [current_time,file_read,file_write,editor,shell,slack]
    
    # MCP Server の設定
    lambda_execute_client = MCPClient(
        lambda: stdio_client(
            StdioServerParameters(
                command="uvx",
                args=[
                    "awslabs.lambda-tool-mcp-server@latest"
                ],
                env={
                    "AWS_PROFILE": "プロファイル名",
                    "FUNCTION_LIST": "r-ide-get-metricsgraph",
                    "AWS_REGION":"ap-northeast-1"
                }
            )
        )
    )
    
    with lambda_execute_client:
        tools.extend(lambda_execute_client.list_tools_sync())
        # エージェントを作成する
        agent = Agent(system_prompt=SYSTEM_PROMPT, tools=tools)
        response = agent("テストケース.mdの「テスト手法」を読み取ってテストを実行してください")
    

    コードは4つの構成要素に分かれています。

    • システムプロンプトの設定
    • 組み込みツールの設定
    • Lambda MCP Serverの設定
    • エージェントの作成

    次章からは、組み込みツールの設定、Lambda MCP Server の設定、そしてエージェントの作成について、それぞれの具体的な実装方法を詳しく解説していきます。

    組み込みツールの設定

    まずは、エージェントにツールを組み込むところから始めましょう。

    ここでいう「ツール」とは、エージェントがさまざまなタスクを実行するための“機能セット”のようなものです。

    Strands Agents では、用途に合わせて選べる多彩なツール群が用意されており、必要に応じて自由に追加・組み合わせることができます。

    https://strandsagents.com/latest/documentation/docs/user-guide/concepts/tools/example-tools-package/strandsagents.com

    今回のプロジェクトで使用したツールと、それぞれの活用方法は以下の通りです。

    • current_time : 現在時刻の取得。ログ解析やメトリクスグラフ取得時の時間指定の際に用いる
    • file_read : テストケースファイルの読み取り
    • file_write : エビデンスファイル作成
    • editor : テストケースファイルへのテスト結果の記入
    • shell : シェルの実行。aws cli を実行する
    • slack : Slackのニュース通知確認

    Lambda MCP Server の設定

    メトリクスグラフ保存において、Lambda MCP Server を経由して、メトリクスグラフ保存用Lambdaを実行できるようにします。

    Lambda MCP Serverとは

    AI が外部ツールとしてAWS Lambda関数を活用できるようにするための仕組み、それが Lambda MCP Server です。

    このサーバーは、MCPクライアント(LLMからの指示を受けて、MCP Server に処理を依頼・結果を受け取る存在)と AWS Lambda関数 の間を橋渡しする役割を担っています。
    自然言語で指示を出すだけで、Lambda関数を呼び出して処理を実行し、その結果を受け取ることが可能になります。

    Lambda MCP Server の概要

    github.com

    今回は、メトリクスグラフの取得をするためのLambda関数を実装し、これをツールとして使用していきます。Lambda 関数についての説明は割愛し、ソースコードだけ載せておきます。

    Lambda関数

    import boto3
    import os
    import datetime
    import json
    from botocore.exceptions import ClientError
    
    def save_cloudwatch_metric_graph(
        metrics_name,
        namespace,
        statistic,
        bucket_name,
        s3_key=None,
        period=300,
        start_time=None,
        end_time=None,
        dimensions=None,
        width=1614,
        height=400
    ):
        """
        CloudWatchメトリクスのグラフを画像としてS3に保存する関数
        
        Parameters:
        -----------
        metrics_name : str
            メトリクス名 (例: 'CPUUtilization')
        namespace : str
            名前空間 (例: 'AWS/EC2')
        statistic : str
            統計 (例: 'Average', 'Sum', 'Maximum', 'Minimum', 'SampleCount')
        bucket_name : str
            S3バケット名
        s3_key : str, optional
            S3オブジェクトキー、デフォルトは自動生成
        period : int
            期間(秒)
        start_time : datetime, optional
            開始時間、デフォルトは1時間前
        end_time : datetime, optional
            終了時間、デフォルトは現在
        dimensions : list, optional
            ディメンション (例: [{'Name': 'InstanceId', 'Value': 'i-1234567890abcdef0'}])
        width : int
            画像の幅
        height : int
            画像の高さ
        y_axis_unit : str, optional
            Y軸の単位 (例: 'Percent', 'Count', 'Bytes', 'Seconds', 'Milliseconds')
        
        Returns:
        --------
        str
            S3のオブジェクトURL
        """
        # 時間範囲の設定
        if end_time is None:
            end_time = datetime.datetime.utcnow()
        if start_time is None:
            start_time = end_time - datetime.timedelta(hours=1)
        
        # AWSクライアントの作成
        cloudwatch = boto3.client('cloudwatch')
        s3 = boto3.client('s3')
        
        # S3キーの生成
        if s3_key is None:
            timestamp = int(datetime.datetime.now().timestamp())
            s3_key = f"cloudwatch-metrics/{metrics_name}_{timestamp}.png"
        
        try:
            # メトリクスの定義
            if dimensions:
                metric_spec = [namespace, metrics_name]
                for dim in dimensions:
                    metric_spec.extend([dim['Name'], dim['Value']])
                metric_spec.append({"yAxis": "left"})
            else:
                metric_spec = [namespace, metrics_name,{"yAxis":"left"}]
            
            # メトリクスウィジェットの定義
            metric_widget = {
                "metrics": [metric_spec],
                "view": "timeSeries",
                "stacked": False,
                "stat": statistic,
                "period": period,
                "title": f"{metrics_name} ({statistic})",
                "width": width,
                "height": height,
                "yAxis":{
                    "left":{
                        "showUnits":True,
                        "min":0,
                        "max":2
                    }
                },
                "start": start_time.strftime("%Y-%m-%dT%H:%M:%S"),
                "end": end_time.strftime("%Y-%m-%dT%H:%M:%S")
            }
    
            # # Y軸の単位設定
            # if y_axis_unit:
            #     metric_widget["yAxis"] = {
            #         "left": {
            #             "unit": y_axis_unit
            #         }
            #     }
            
            # メトリクスウィジェットイメージの取得
            import json
            response = cloudwatch.get_metric_widget_image(
                MetricWidget=json.dumps(metric_widget)
            )
            
            # 画像データの取得
            image_data = response['MetricWidgetImage']
            
            # S3に画像をアップロード
            s3.put_object(
                Bucket=bucket_name,
                Key=s3_key,
                Body=image_data,
                ContentType='image/png'
            )
            
            s3_url = f"s3://{bucket_name}/{s3_key}"
            print(f"メトリクスグラフをS3に保存しました: {s3_url}")
            return s3_url
        
        except ClientError as e:
            print(f"エラーが発生しました: {e}")
            raise
    
    def lambda_handler(event, context):
        """
        Lambda関数のハンドラー
        
        Parameters:
        -----------
        event : dict
            Lambda関数のイベントデータ。Cloudwatchのメトリクス名や名前空間、メトリクスグラフの保存先S3バケット名などのパラメータが含まれる。
            予想される形式は、例えば、{"metrics_name": "Invocations"}や{"namespace":"AWS/Lambda"}など
        context : object
            Lambda関数のコンテキスト
        
        Returns:
        --------
        dict
            Lambda関数のレスポンス
        """
        try:
            print(f"受信したイベント: {event}")
            # イベントからパラメータを取得(または固定値を使用)
            metrics_name = event.get('metrics_name')
            namespace = event.get('namespace')
            statistic = event.get('statistic')
            period = event.get('period',60)
            bucket_name = event.get('bucket_name')  # 必須パラメータ
            s3_key = event.get('s3_key', None)
            
            if not bucket_name:
                raise ValueError("bucket_nameパラメータが必要です")
            
            # 時間範囲の設定
            hours = event.get('hours', 1)
            end_time = datetime.datetime.utcnow()
            start_time = end_time - datetime.timedelta(hours=hours)
    
            # ディメンションの設定
            dimensions = event.get('dimensions')
            
            # グラフのサイズ設定
            width = event.get('width', 1614)
            height = event.get('height', 386)
    
            # Y軸の単位設定
            y_axis_unit = event.get('y_axis_unit')
            
            # グラフの保存
            s3_url = save_cloudwatch_metric_graph(
                metrics_name=metrics_name,
                namespace=namespace,
                statistic=statistic,
                bucket_name=bucket_name,
                s3_key=s3_key,
                period=period,
                start_time=start_time,
                end_time=end_time,
                dimensions=dimensions,
                width=width,
                height=height,
            )
            
            return {
                'statusCode': 200,
                'body': f'メトリクスグラフをS3に保存しました: {s3_url}',
                's3_url': s3_url
            }
        
        except Exception as e:
            return {
                'statusCode': 500,
                'body': f'エラーが発生しました: {str(e)}'
            }
    

    各パラメーターについては、以下の記事が参考になるかと思います。

    tech.nri-net.com

    注意点として、AI に関数が受け取る入力データ(イベント)を明確に定義・認識させる必要があります。
    システムプロンプト内に、以下のように、Lambda 関数で受け取るデータ型を記載するようにしましょう。

    <注意点>
     AWS Lambda関数(r-ide-get-metricsgraph)を実行する際に、テストケースファイル内のパラメータとLambda関数の入力パラメータが一致している必要があります。
     テストケースのパラメータをLambda関数の期待する形式に変換・調整するようにしてください。なお、Lambda関数は以下のようなパラメータを期待しています。
     {
      "metrics_name": "メトリクス名",
      "namespace": "名前空間",
      "statistic": "Average",
      "period": 60,
      "hours": 24,
      "bucket_name": "S3バケット名",
      "dimensions": [
        {
          "Name": "ディメンジョンの名前",
          "Value": "ディメンジョンの値"
        }
      ]
    }
    

    <補足事項>

    • 今回の検証時点では、CloudWatch MCP Server は未提供だったため、検証には含めていません。ただし、現在では今回の実装と同様のことが可能になっている可能性もあります。最新の情報については、公式ドキュメントやアップデートを確認してください。

    github.com

    • 組み込みツールであるuse_awsを使わなかったのは、不要な権限を持たせたくなかったという理由があります。具体的には、CloudWatchのログ閲覧に限定したIAMロールをLambda に設定することで、最小権限の原則を実現しています。

    MCP Clientを使って Lambda ツールに接続する

    では、Lambda MCP Server と接続していきます。

    MCP Client からツール一覧を取得する流れ

    • ソースコードの説明
      • MCPClientの中にMCP Server の設定を入れる
      • envの中には環境変数を設定する
      • AWS_POFILEにはAWS CLIで設定したプロファイル名を記載する
      • FUNCTION_LISTにはLambda 関数名を指定する
      • list_tools_sync()を用いることで、必要なツール一覧を取得できる

    エージェントの作成

    最後に、エージェントの構築を行います。これまでに定義したシステムプロンプトや追加したツールを統合し、目的に応じた動作が可能なエージェントを完成させます。

    with lambda_execute_client:
    # Agent の作成
        agent = Agent(system_prompt=SYSTEM_PROMPT,tools=tools) #システムプロンプト、ツールを設定する。
        response = agent("テストケース.mdの「テスト手法」を読み取ってテストを実行してください")
    

    いざ実行

    pythonファイルを以下のコマンドで実行します。

    uv run agent.py
    

    結果を見てみましょう。

    結果はこちら

    テストケースファイルを読み取り、テスト手法に従ってテストを実行します。まずはテストケースファイルの内容を確認しましょう。
    Tool #1: file_read
    まず、「エビデンス」フォルダを確認して、必要であれば作成します。
    Tool #2: shell
    Do you want to proceed with execution? [y/*] y
    テストケースファイルから、複数のテストケースが確認できました。それぞれのテスト手法に基づいて順番に実行していきます。
    
    ### テストケース1: 定期実行の確認 (AWS/EVENTS)
    Tool #3: current_time
    まず、AWS CLIを使用してEventBridgeの定期実行のメトリクスを確認します。
    Tool #4: shell
    Do you want to proceed with execution? [y/*] y
    {
        "Label": "Invocations",
        "Datapoints": [
            {
                "Timestamp": "2025-07-12T01:00:00+00:00",
                "Average": 1.0,
                "Unit": "Count"
            }
        ]
    }
    次に、メトリクスグラフを取得するためにLambda関数を呼び出します:
    Tool #5: r_ide_get_metricsgraph
    INFO:mcp.server.lowlevel.server:Processing request of type CallToolRequest
    S3バケットからグラフをダウンロードします:
    Tool #6: shell
    Do you want to proceed with execution? [y/*] y
    download: s3://r-ide-mcp-test-bucket/cloudwatch-metrics/Invocations_1752284839.png to エビデンス/テストケース1_AWS_Events_Invocations.png
    結果をエビデンスファイルにまとめます:
    Tool #7: file_write
    Do you want to proceed with the file write? [y/*] y
    ### テストケース2: アプリログの確認(ログの表示)
    
    CloudWatch Logs Insightsを使用してログを確認します:
    Tool #8: shell
    Do you want to proceed with execution? [y/*] y
    {
        "queryId": "3251cd4f-38aa-453e-aa44-c6b5272b3752"
    }
    クエリ結果を取得します:
    Tool #9: shell
    Do you want to proceed with execution? [y/*] y
    {
        "queryLanguage": "CWLI",
        "results": [
            [
                {
                    "field": "@timestamp",
                    "value": "2025-07-12 01:00:22.163"
                },
                {
                    "field": "@message",
                    "value": "END RequestId: dad7234c-11a3-450d-9637-cfc31da6d4d8\n"
                },
      
    クエリの結果をもう少し取得します:
    Tool #10: shell
    Do you want to proceed with execution? [y/*] y
    {
        "queryLanguage": "CWLI",
        "results": [
            [
                {
                    "field": "@timestamp",
                    "value": "2025-07-12 01:00:22.163"
                },
                {
                    "field": "@message",
                    "value": "END RequestId: dad7234c-11a3-450d-9637-cfc31da6d4d8\n"
                },
                {
                    "field": "@ptr",
                    ~~~~省略~~~~
         
    さらに全体の結果を確認してみます:
    Tool #11: shell
    Do you want to proceed with execution? [y/*] y
    {
        "queryLanguage": "CWLI",
        "results": [
            [
                {
                    "field": "@timestamp",
                    "value": "2025-07-12 01:00:22.163"
                },
                {
                    "field": "@message",
                    "value": "END RequestId: dad7234c-11a3-450d-9637-cfc31da6d4d8\n"
                },
                {
                    "field": "@ptr",
    クエリ結果の一部を確認できました。エビデンスファイルに結果をまとめます:
    Tool #12: file_write
    Do you want to proceed with the file write? [y/*] y
    ### テストケース3: アプリログの確認(ERRORログがないことを確認)
    Tool #13: shell
    Do you want to proceed with execution? [y/*] y
    {
        "queryId": "d60cdbb9-e1ef-4500-af7b-8c03c124a89c"
    }
    クエリ結果を取得します:
    Tool #14: shell
    Do you want to proceed with execution? [y/*] y
    {
        "queryLanguage": "CWLI",
        "results": [],
        "statistics": {
            "recordsMatched": 0.0,
            "recordsScanned": 26.0,
            "estimatedRecordsSkipped": 0.0,
            "bytesScanned": 4482.0,
            "estimatedBytesSkipped": 0.0,
            "logGroupsScanned": 1.0
        },
        "status": "Complete"
    }
    テスト結果をエビデンスファイルにまとめます:
    Tool #15: file_write
    Do you want to proceed with the file write? [y/*] y
    ### テストケース4: 定期実行の確認 (AWS/Lambda Invocations)
    Tool #16: shell
    Do you want to proceed with execution? [y/*] y
    {
        "Label": "Invocations",
        "Datapoints": [
            {
                "Timestamp": "2025-07-12T01:00:00+00:00",
                "Average": 1.0,
                "Unit": "Count"
            }
        ]
    }
    次に、メトリクスグラフを取得します:
    Tool #17: r_ide_get_metricsgraph
    INFO:mcp.server.lowlevel.server:Processing request of type CallToolRequest
    S3バケットからグラフをダウンロードします:
    Tool #18: shell
    Do you want to proceed with execution? [y/*] y
    download: s3://r-ide-mcp-test-bucket/cloudwatch-metrics/Invocations_1752285010.png to エビデンス/テストケース4_AWS_Lambda_Invocations.png
    結果をエビデンスファイルにまとめます:
    Tool #19: file_write
    Do you want to proceed with the file write? [y/*] y
    ### テストケース5: 定期実行の確認 (AWS/Lambda Errors)
    Tool #20: shell
    Do you want to proceed with execution? [y/*] y
    {
        "Label": "Errors",
        "Datapoints": [
            {
                "Timestamp": "2025-07-12T01:00:00+00:00",
                "Average": 0.0,
                "Unit": "Count"
            }
        ]
    }
    メトリクスグラフを取得します:
    Tool #21: r_ide_get_metricsgraph
    INFO:mcp.server.lowlevel.server:Processing request of type CallToolRequest
    S3バケットからグラフをダウンロードします:
    Tool #22: shell
    Do you want to proceed with execution? [y/*] y
    download: s3://r-ide-mcp-test-bucket/cloudwatch-metrics/Errors_1752285057.png to エビデンス/テストケース5_AWS_Lambda_Errors.png
    結果をエビデンスファイルにまとめます:
    Tool #23: file_write
    Do you want to proceed with the file write? [y/*] y
    ### テストケース6: Slack通知の確認
    
    Slackの通知を確認します:
    Tool #24: slack
    スレッドの内容も確認しましょう:
    Tool #25: slack
    タイムスタンプを確認しましょう:
    Tool #26: shell
    Do you want to proceed with execution? [y/*] y
    2025-07-12 10:00:19 JST
    結果をエビデンスファイルにまとめます:
    Tool #27: file_write
    Do you want to proceed with the file write? [y/*] y
    ### テスト結果のまとめと「テストケース.md」の更新
    
    テスト結果をまとめます:
    Tool #28: editor
    すべてのテスト結果に基づいて、テストケースファイルを更新します:
    Tool #29: editor
    Do you want to proceed with the str_replace operation? [y/*] y
    テスト結果をまとめたファイルを作成します:
    Tool #30: file_write
    Do you want to proceed with the file write? [y/*] y
    テストケースの実行と結果の確認が完了しました。すべてのテストケースで「OK」という結果が得られました。
    
    ## 実施したテストの概要:
    
    1. **AWS EventBridgeの定期実行確認**:
       - AWS/EVENTSネームスペースのInvocationsメトリクスを確認し、午前10時頃(JST)に実行されていることを確認しました
       - メトリクスグラフを取得してエビデンスとして保存しました
    
    2. **Lambda関数のログ確認**:
       - CloudWatch Logs Insightsを使用して関数のログを確認し、正常に動作していることを確認しました
       - エラーログが存在しないことも確認しました
    
    3. **Lambda関数のメトリクス確認**:
       - AWS/Lambdaネームスペースの以下のメトリクスを確認しました:
         - Invocations: 午前10時頃(JST)に1回以上実行されていることを確認
         - Errors: 午前10時頃(JST)にエラーが0であることを確認
    
    4. **Slack通知の確認**:
       - 指定されたSlackチャンネル「times-r-ide」にて午前10時頃(JST)にAWSニュースが通知されていることを確認しました
       - 通知内容や時間を確認しエビデンスとして記録しました
    
    すべてのテスト結果とエビデンスは「エビデンス」フォルダに保存され、テストケース.mdファイルも更新されました。テスト結果のまとめレポートも作成し、全テストケースが合格(OK)していることが確認できました。

    テストケースファイル

    更新されたテストケースファイル

    エビデンスファイルの例

    エビデンスファイルの例

    ログの取得から可視化、結果の記録に至るまで、Agentがツールを用いてタスクを実行している様子が読み取れます。

    こんなこともできる

    ここまでは一つのエージェントのみでタスクを実行してきました。
    しかし、テストケースが複雑化し、数も増えてくると、ひとつのエージェントだけでは対応が難しくなる場面も出てきます。

    そこで活用したいのが、Strands Agents の「マルチエージェント機能」です。

    マルチエージェントの概要

    この機能では、以下のような構成が可能になります:

    • コーディネーターエージェント:タスクを受け付け、内容に応じて適切なエージェントに振り分ける役割を担います。

    • 専門家エージェント群:それぞれが特定の分野や処理に特化しており、コーディネーターから割り当てられたタスクを効率的に処理します。

    この構図により、タスクの分散処理が可能になり、スケーラビリティと保守性が大幅に向上します。特に、テストケースの種類が多岐にわたるプロジェクトでは、マルチエージェント構成が非常に有効です。

    実装してみる

    では、先ほどのソースコードをマルチエージェント用に書き替えてみましょう。
    まずは専門家エージェント群の実装からです。

    専門家エージェント群の実装

    専門家エージェントは@tool デコレーターを用いて実装していきます。
    この機能を使えば、開発者自身が特定の処理やロジックをツールとして定義し、エージェントから呼び出せるようになります。これにより、エージェントの構成をより分かりやすく整理でき、機能の再利用性や保守性が向上します。

    @tool を用いたマルチエージェントの実装

    さらに、ツールの内部にエージェントを組み込むことも可能です。これにより、ツールが単なる関数的な処理だけでなく、エージェントの知識や対話能力を活かした高度な処理を実現することができます。

    ソースコードは以下の通り

    specialized_agents.py

    from strands import Agent, tool
    from strands_tools import current_time, file_write,file_read,editor,current_time,shell,slack
    from strands.tools.mcp import MCPClient
    from mcp import stdio_client, StdioServerParameters
    
    # Define a specialized system prompt
    RESEARCH_ASSISTANT_PROMPT = """
    あなたはAWS CLIを実行して、メトリクス分析、ログの分析を実施する専門エージェントです。
    orchestratorエージェントから渡された情報を基に、aws cliを実行してください。
    - テストを実行すると同時にエビデンス取得を行う必要があります。取得したエビデンスはmarkdownファイル形式で
    「エビデンス」フォルダに格納してください。
    - エビデンスには、検索結果だけではなく、検索結果の根拠として、検索に使った具体的なaws cliのコマンドを表示してください。
    """
    
    SLACK_PROMPT="""
    あなたはSlack通知の確認を担当するAIエージェントです。
    orchestratorエージェントから渡された情報を基に、指定されたSlackチャンネルの通知を確認してください:
    
    - チャンネル名
    - チャンネルID
    
    <注意点>
    - エビデンスは該当スレッドのURLと通知の内容をmarkdownファイルにまとめるようにしてください。
    - エビデンスは「エビデンス」フォルダに格納するようにしてください。
    """
    
    Lambda_PROMPT="""
    あなたは、メトリクスグラフ保存Lambdaを実行するためのエージェントです。
    orchestratorエージェントから渡された情報を基にLambda MCP Server経由でLambda関数r-ide-get-metricsgraphを実行してください
    
    <注意点>
     AWS Lambda関数(r-ide-get-metricsgraph)を実行する際に、テストケースファイル内のパラメータとLambda関数の入力パラメータが一致している必要があります。
     パラメータをLambda関数の期待する形式に変換・調整するようにしてください。なお、Lambda関数は以下のようなパラメータを期待しています。
     {
      "metrics_name": "メトリクス名",
      "namespace": "名前空間",
      "statistic": "Average",
      "period": 60,
      "hours": 24,
      "bucket_name": "r-ide-mcp-test-bucket",
      "dimensions": [
        {
          "Name": "ディメンジョンの名前",
          "Value": "ディメンジョンの値"
        }
      ]
    }
    また、以下二つのエビデンス取得作業を実施してください。
        - aws cliを用いて、そのメトリクスが何時に発生したかを記載する。また、実行したaws cliのコマンドをエビデンスファイルに記載する。
        - S3バケットからCLIを使って「エビデンス」フォルダーにメトリクスグラフをダウンロードする
    """
    @tool
    def cli_execute_agent(query: str) -> str:
        try:
            # Strands Agents SDK makes it easy to create a specialized agent
            cli_agent = Agent(
                system_prompt=RESEARCH_ASSISTANT_PROMPT,
                tools=[current_time,shell,file_write,editor]  # Research-specific tools
            )
    
            # Call the agent and return its response
            response = cli_agent(query)
            return str(response)
        except Exception as e:
            return f"Error in research assistant: {str(e)}"
    @tool
    def slack_execute_agent(query: str) -> str:
        try:
            slack_agent = Agent(
                tools=[slack,file_write,shell,current_time,editor],  # Tools for getting product data
            )
            # Implementation with response handling
            response = slack_agent(query)
            processed_response = f"テストの実行を始めます:\n{response}"
            return processed_response
        except Exception as e:
            return f"Error in product recommendation: {str(e)}"
    
    @tool
    def lambda_execute_agent(query: str) -> str:
        try:
            lambda_execute_client = MCPClient(
            lambda: stdio_client(
                StdioServerParameters(
                    command="uvx",
                    args=[
                        "awslabs.lambda-tool-mcp-server@latest"
                    ],
                    env={
                        "AWS_PROFILE": "プロファイル名",
                        "FUNCTION_LIST": "r-ide-get-metricsgraph",
                        "AWS_REGION":"ap-northeast-1"
                    }
                )
            )
        )
            with lambda_execute_client:
                tools = lambda_execute_client.list_tools_sync()
                tools.append(cli_execute_agent)
                lambda_agent = Agent(
                    system_prompt=Lambda_PROMPT, 
                    tools=tools
                )
            # Implementation with response handling
            # ...
                response = lambda_agent(query)
                processed_response = f"Lambdaを実行します:\n{response}"
                return processed_response
        except Exception as e:
            return f"Error in trip planning: {str(e)}"
    

    コーディネーターエージェントの実装

    次に、コーディネーターエージェントの実装です。@toolでツール化した専門家エージェント群をツールとして追加していきます。

    コーディネーターエージェントの実装

    multi_agent.py

    from strands import Agent
    from strands_tools import file_read,editor
    from specialized_agents import cli_execute_agent, slack_execute_agent, lambda_execute_agent
    
    user_message = input("テストケースファイル名: ")
    # Define the orchestrator system prompt with clear tool selection guidance
    MAIN_SYSTEM_PROMPT = f"""
    あなたはAWSリソースの総合テストを管理するAIエージェントです。
    {user_message}を読み取り、テスト手法を判断してください。
    ファイルの「テスト手法」列には、以下のような情報が含まれています:
        - テストの手法
        - テストを実施するときに必要なパラメーター情報
            - メトリクスグラフ、履歴を取得するためのパラメータ(例:namespace、metrics_name,period)
            - Logs Insights を実行するためのパラメータ(例:ロググループ名、クエリ)
            - slackのチャンネル名
    
    取得した情報をもとに、以下のエージェントにタスクを振り分けてください。この際、「テスト手法」列の情報を各エージェントに渡すようにしてください
    
    - テスト手法が「メトリクスの分析」の場合 → lambda_execute_agent
    - テスト手法が「Slack通知の確認」→ slack_execute_agent
    - テスト手法が「CloudWatchのログのインサイトを使用する」 → cli_execute_agent
    
    各エージェントから得られた結果と「期待される結果」を比較し、{user_message}にいずれかを記録してください:
        - 条件を満たしていれば:「テスト結果」列に"OK"を記入する
        - 条件を満たしていなければ:「テスト結果」列に"NG"を記入する
    
    """
    
    orchestrator = Agent(
        system_prompt=MAIN_SYSTEM_PROMPT,
        tools=[editor,file_read,cli_execute_agent, slack_execute_agent, lambda_execute_agent]
    )
    response = orchestrator(user_message)
    

    いざ実行

    以下commandで実行します。

    uv run multi-agent.py
    

    するとユーザー入力が求められます。ここにテストケースファイルの名前を入力しましょう。

    テストケースファイル名: {入力部分}
    

    結果はこちらです。(長すぎるので一部抜粋としました)

    Tool #1: file_read
    テストケース.mdファイルを確認しました。これからテストケース.mdのテスト手法に従って、適切なエージェントにタスクを振り分け、テスト結果を記録していきます。
    
    まず、各テストケースの情報を整理し、適切なエージェントにタスクを振り分けます。
    
    ### テストケース1: 定期実行の確認
    - テスト手法: メトリクスを分析
    - パラメータ: NameSpace: AWS/EVENTS, Metrics_NAME:Invocations, EventRule:r-ide-get-aws-news, 期間:直近1日, period:60
    - 期待される結果: 10時(JST)にInvocationsが1以上
    
    ########## Lambda 専門エージェントの呼び出し###########(この文は筆者が追記)
    
    Tool #2: lambda_execute_agent
    AWS EventBridgeのメトリクスを分析するためのリクエストを承りました。指定された以下の条件に基づいてLambda関数を実行し、メトリクスを確認します:
    
    - Namespace: AWS/EVENTS
    - Metrics Name: Invocations
    - Dimension: EventRule = r-ide-get-aws-news
    - 期間: 直近1日
    - 集計間隔: 60秒
    
    Lambda関数を実行して必要なメトリクスデータを取得します。
    Tool #1: r_ide_get_metricsgraph
    INFO:mcp.server.lowlevel.server:Processing request of type CallToolRequest
    メトリクスグラフを正常にS3バケットに保存しました。次に、JST 10時におけるInvocationsが1以上かどうかを確認するための調査を行います。
    
    AWS CLIを使ってメトリクスデータを詳細に分析します:
    
    ########## CLI 専門エージェントの呼び出し###########(この文は筆者が追記)
    
    Tool #2: cli_execute_agent 
    AWS CLIを使用して、AWS/EVENTSネームスペースのInvocationsメトリクスを取得します。RuleName=r-ide-get-aws-newsのディメンションに対して、最近24時間のJST 10時前後のデータを確認します。
    
    まず、現在のUTC時間を確認し、JST(UTC+9)との時差を考慮してコマンドを実行します。
    Tool #1: current_time
    現在のUTC時間は2025-07-13 09:21:30です。JST(日本時間)は UTC+9 なので、JST 10:00はUTC 01:00に相当します。
    
    それでは、AWS/EVENTSネームスペースのInvocationsメトリクスを取得するコマンドを実行します。JSTの10時前後のデータを取得するため、UTC時間では01:00前後(00:00~02:00)のデータを取得します。最近24時間のデータを取得するため、期間を24時間に設定します。
    Tool #2: shell
    Do you want to proceed with execution? [y/*] y
    {
        "Label": "Invocations",
        "Datapoints": [
            {
                "Timestamp": "2025-07-13T01:00:00+00:00",
                "Average": 1.0,
                "Unit": "Count"
            }
        ]
    }

    ログからも分かる通り、Lambda 専門エージェントと CLI 専門エージェントが呼び出されて、タスクが実行されていることが分かります。

    シングルエージェントとの違い

    今回のテスト自動化では、マルチエージェント構成を採用することで、より詳細な分析が可能になりました。
    たとえば、各テストケースに対して個別に「テスト結果サマリー」をログに表示するなど、より詳細な調査をしてくれました。

    テスト結果サマリー

    ## 調査結果サマリー
    
    1. **AWS Lambda Invocationsメトリクスの確認**
       - Lambda関数: r-ide-get-aws-news
       - 時間帯: 日本時間10時 (2025-07-13T01:00:00+00:00 UTC)
       - Invocations値: **1.0**
       - 判定: 「1以上かどうか」という確認に対して **YES** (1以上である)
    
    2. **S3バケットからのメトリクスグラフダウンロード**
       - ソース: s3://r-ide-mcp-test-bucket/cloudwatch-metrics/Invocations_1752398852.png
       - 保存先: エビデンス/Invocations_1752398852.png
       - 結果: 正常にダウンロード完了(ファイルサイズ約22.6KB)
    
    すべての調査結果は「エビデンス/AWS_Lambda_Invocations分析結果.md」に詳細に記録されています。
    
    何か他に確認や追加の調査が必要であればお知らせください。# AWS Lambda Invocationsメトリクス分析結果
    
    ## 分析概要
    - **NameSpace**: AWS/Lambda
    - **FunctionName**: r-ide-get-aws-news
    - **Metrics_name**: Invocations
    - **期間**: 直近1日間
    - **集計間隔**: 60秒
    
    ## 分析結果
    
    ### メトリクス値の確認
    AWS CLIを使用して10時(JST)のInvocationsメトリクスを確認した結果、**Invocationsの値は1.0**であることが確認できました。つまり、10時(JST)において、r-ide-get-aws-news Lambda関数は1回呼び出されていることが判明しました。
    
    ### 実行したコマンド
    
    aws cloudwatch get-metric-statistics \
    --namespace AWS/Lambda \
    --metric-name Invocations \
    --dimensions Name=FunctionName,Value=r-ide-get-aws-news \
    --start-time 2025-07-13T01:00:00Z \
    --end-time 2025-07-13T02:00:00Z \
    --period 60 \
    --statistics Average
    
    ### エビデンス収集
    1. メトリクスグラフが正常にS3バケットに保存されました:
       - 保存先: `s3://r-ide-mcp-test-bucket/cloudwatch-metrics/Invocations_1752398852.png`
    
    2. グラフ画像を「エビデンス」フォルダにダウンロードしました:
       - ダウンロード先: `エビデンス/Invocations_1752398852.png`
       - 使用したコマンド:
         ```bash
         aws s3 cp s3://r-ide-mcp-test-bucket/cloudwatch-metrics/Invocations_1752398852.png エビデンス/
         ```
    
    ## 結論
    10時(JST)において、r-ide-get-aws-news Lambda関数のInvocations値は1.0であり、**Invocationsが1以上**という条件を満たしていることが確認できました。

    一方、マルチエージェント構成は各々のエージェントが詳細な分析を行うため、処理時間が長くなる傾向があります。 そのため、シンプルなタスクや、短い処理時間が求められるケースでは、マルチエージェント構成は適さないと言えるでしょう。

    シングルエージェントとマルチエージェントの使い分けについて、明確な境界線があるわけではありません。
    プロジェクトの規模、求められる精度、処理時間とのバランスを見ながら、状況に応じて柔軟に選択することが重要です。

    まとめ

    本記事では、Strands Agents を活用したテスト自動化の取り組みについてご紹介しました。
    実際に試行した結果、一定の効果や可能性を感じられた一方で、現時点では本ツールをそのまま実プロジェクトに導入するには至っていません。
    その理由の一つは、セキュリティ面の配慮や適切な権限設定の必要性です。(他にも、エージェントの運用・監視方法など、検討すべき課題は多くあります)
    特に企業環境では、これらの要素がクリアされていないと導入は難しいと言わざるを得ません。

    そういう意味でも、今後は Amazon Bedrock AgentCore を活用し、より安全かつ実用的なツールの実装を目指していきたいと考えています。

    Bedrock Agent Core に関しては以下の記事をご覧ください!

    tech.nri-net.com

    執筆者:井手 亮太
    職種:インフラエンジニア
    推しのサッカー選手:ケビン・デブライネ
    執筆記事一覧:https://tech.nri-net.com/archive/author/r-ide-ryota