AmazonBedrockとPlayWrightMCPで始める日本語E2Eテスト #Python – Qiita

from datetime import datetime, timedelta
from mcp import stdio_client, StdioServerParameters
from pathlib import Path
from strands import Agent
from strands.handlers.callback_handler import CompositeCallbackHandler, PrintingCallbackHandler
from strands.models import BedrockModel
from strands.tools.mcp import MCPClient

import boto3
import logging
import os
import shutil
import threading


# Enables Strands debug log level
logger = logging.getLogger("strands")
logger.setLevel(logging.INFO)

finished = threading.Event() 

os.environ["DEBUG"]="pw:browser*"
os.environ["HOME"]="/tmp"


def get_chromium_executable_path() -> str:
    """ 
    PlaywrightのChromium実行ファイルのパスを取得する関数
    chromium-{バージョン番号}に実行ファイルが保存されるため、今回のコンテナで駆動している番号を動的に取得する必要がある。
    :return: Chromium実行ファイルのパス
    """
    base_path_str = "/opt/.cache/ms-playwright/"
    chromium_dir = next(iter(Path(base_path_str).glob("chromium-*")))

    return chromium_dir.joinpath("chrome-linux", "chrome").as_posix()


def upload_screenshot_to_s3(screenshots_dir: str, output_s3_bucket_uri: str):
    """ S3にスクリーンショットをアップロードする関数
    :param screenshots_dir: スクリーンショットを保存するディレクトリ
    :param output_s3_bucket_uri: S3のバケットURI
    :return: None
    """
    # スクリーンショットが存在する場合、S3にアップロード
    logger.info("Checking for screenshots to upload. screenshots-dir: %s Output S3 bucket URI: %s", screenshots_dir, output_s3_bucket_uri)
    s3_client = boto3.client('s3')
    

    # スクリーンショットディレクトリが存在しない場合は何もしない
    if not os.path.exists(screenshots_dir):
        logger.info("No screenshots directory found at %s", screenshots_dir)
        return
    logger.info("scrrenshots_dir files: %s", os.listdir(screenshots_dir))
    # スクリーンショットディレクトリが存在する場合は、ファイルをS3にアップロード
    for file_name in os.listdir(screenshots_dir):
        if file_name.endswith(".png") or file_name.endswith(".jpg") or file_name.endswith(".jpeg"):
            file_path = os.path.join(screenshots_dir, file_name)
            s3_key = f"{file_name}"
            bucket_name, *folder_path = output_s3_bucket_uri.split("s3://")[1].split("/", 1)
            s3_client.upload_file(file_path, bucket_name, "".join(folder_path + [s3_key]))
            logger.info(f"Uploaded {file_path} to s3://{output_s3_bucket_uri.split('s3://')[1]}/{s3_key}")    
    return



def lambda_handler(event, context):
    
    ####################################################
    # 事前準備
    ####################################################

    # playwrightパスを取得する
    playwright_browser_path = get_chromium_executable_path()

    # 今回のサンプルで設定するテストイベントのキーの有無をチェック
    if event.get("access_url") is None or event.get("output_s3_bucket_uri") is None or event.get("text_task") is None:
        raise ValueError("Missing required parameters: access_url, output_s3_bucket_uri, text_task")
    access_url = event.get("access_url")
    output_s3_bucket_uri = event.get("output_s3_bucket_uri")
    text_task = event.get("text_task")
    bedrock_model_id = event.get("bedrock_model_id", "anthropic.claude-3-5-sonnet-20241022-v2:0")

    # 現在の日時を取得（JST: UTC+9）
    current_time = (datetime.utcnow() + timedelta(hours=9)).strftime("%Y-%m-%dT%H:%M:%S")



    # スクリーンショット用の現在のディレクトリ配下に作成する。
    screenshots_dir = "/tmp/screenshots"
    if not os.path.exists(screenshots_dir):
        os.makedirs(screenshots_dir)
        logger.info("Created screenshots directory: %s", screenshots_dir)
    # スクリーンショットディレクトリが存在する場合
    else:
        # 既存のスクリーンショットディレクトリをリセット
        
        shutil.rmtree(screenshots_dir)
        os.makedirs(screenshots_dir)
        logger.info("Reset and created new screenshots directory: %s", screenshots_dir)



    ####################################################
    # MCPクライアントの設定とエージェントの実行
    ####################################################

    # 利用する市中のMCPサーバーのクライアントを作成
    playWright_client = MCPClient(
        lambda: stdio_client(StdioServerParameters(command="npx", args=["-y", "@playwright/[email protected]", "--config", "/var/task/mcp.config.json" ,"--executable-path" ,f"{playwright_browser_path}", "--output-dir", f"{screenshots_dir}"], environ=os.environ))
        
    )

    # Bedrockモデルの定義
    bedrock_model = BedrockModel(
        model_id=bedrock_model_id,
        temperature=0.3,
        streaming=True,
    )

    


    text_prompt = f"""
    以下のアプリに対して、taskの内容を実行してください。 
    ## target
    {access_url}

    ## rule
    - If you are using PlayWright, ensure that headless: true is always set.
    -作業をする際には 下記順番で必ず実行してください。
        - browser_navigate → browser_capture_snapshot → browser_resize → browser_take_screenshot
        - スクリーンショットを撮影する場合のキー名のprefixとして{current_time}を使用してください。
        - スクリーンショットを撮影する場合はbrowser_take_screenshotを使用し、その引数filenameに上記キー名を指定すること。


    ## task
    {text_task}

    ## Output 
    エラーが発生した際にはそのエラーログの原文をそのまま出力してください。読み上げて下さい。
    - 各ケースの結果をOK/NGで報告してください
    - テスト結果がNGの場合、失敗した場合は簡潔な理由を添えてください
    - すべてのテスト結果をテーブル表示で、サマリを表示してください
    
    """
    


    

    ######################################################
    # Playwright MCPクライアントを使用してエージェントを実行
    ######################################################
    with playWright_client:
        # Agentを作成し、ツール(作業タスク)を登録
        agent = Agent(tools=playWright_client.list_tools_sync(), model=bedrock_model)
        response = agent(text_prompt, stream=True)
        finished.wait(timeout=20)
        logger.info("Agent response: %s", response)

        upload_screenshot_to_s3(
            screenshots_dir=screenshots_dir,
            output_s3_bucket_uri=output_s3_bucket_uri,
        )
        logger.info("self printing. Screenshots uploaded to S3 bucket: %s", output_s3_bucket_uri)
        
    logger.info("Lambda handler completed successfully.")
    return