[Amazon Bedrock AgentCore] AgentCore Browser をAIエージェントで操作してみた

from typing import Dict, Any
import os
import logging

from strands import Agent, tool
from strands.models import BedrockModel

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from bedrock_agentcore.tools.browser_client import BrowserClient
from browser_use import Agent as BrowserUseAgent
from browser_use.browser.session import BrowserSession as BU_BrowserSession
from browser_use.browser import BrowserProfile as BU_BrowserProfile
from langchain_aws import ChatBedrockConverse
from playwright._impl._errors import Error as PlaywrightError

os.environ['UVICORN_TIMEOUT_KEEP_ALIVE'] = '28800'
os.environ['UVICORN_TIMEOUT_GRACEFUL_SHUTDOWN'] = '28800'

app = BedrockAgentCoreApp()

# ログ設定（LOG_LEVEL 環境変数で制御: DEBUG/INFO/WARNING/ERROR）
_log_level = os.environ.get("LOG_LEVEL", "INFO").upper()
logging.basicConfig(
    level=getattr(logging, _log_level, logging.INFO),
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("browser_agent")

import contextlib

SYSTEM_PROMPT = """あなたはWeb自動化のアシスタントです。

原則:
1. ユーザーの指示を正確に読み取ってください
2. Web上の操作はrun_browser_taskツールを使って実行してください
3. 実施した操作と結果を簡潔に説明してください
4. CAPTCHAなど人間による検証が必要な場合は、次に取るべき行動を明示してください

ツールの使い方:
- 検索が必要な場合は、ユーザー入力から「短い検索語（名詞句）」を抽出してください。
- use_browserツールを呼び出す際、instruction引数には検索語のみを渡してください（例: 「スーパーマーケット」）。
- 説明文や敬語、語尾（〜を検索してください 等）は渡さないでください。
 - サイト内検索を行う場合は、ページ内の検索テキストボックスと検索ボタンを必ず使用し、アドレスバーや外部検索エンジンは使用しないでください。
 - 例: Yahoo! JAPANトップページ（https://www.yahoo.co.jp/）では、ページ内の検索ボックスに直接入力し、同ページの検索ボタンで検索を実行してください（外部検索は禁止）。

常に簡潔で有用な結果を返してください。
"""

@tool
async def run_browser_task(instruction: str, starting_page: str = "https://www.yahoo.co.jp/") -> str:
    """Execute web automation steps using AgentCore Browser and Browser-Use SDK.

    instruction: Natural language instruction (e.g., "macbookを検索して最初の商品の詳細を抽出してください").
    starting_page: Initial URL to open.
    """

    logger.info("use_browser start: starting_page=%s", starting_page)

    client = BrowserClient(region="us-west-2")
    bu_session = None
    try:
        client.start()
        ws_url, headers = client.generate_ws_headers()
        logger.info("browser session created (region=us-west-2)")
        logger.info("cdp ws_url: %s", ws_url[:100] + "..." if len(ws_url) > 100 else ws_url)

        profile = BU_BrowserProfile(
            headers=headers,
            timeout=180000,
        )
        bu_session = BU_BrowserSession(
            cdp_url=ws_url,
            browser_profile=profile,
        )

        logger.info("Starting browser session...")
        await bu_session.start()
        logger.info("Browser session started successfully")

        bedrock_chat = ChatBedrockConverse(
            model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
            region_name="us-west-2"
        )

        query = instruction.strip() if instruction else "ラ・ムー"
        task = (
            f"最初に、ブラウザツールの検索/URLバーに『{starting_page}』を入力してYahoo! JAPANトップへ移動してください。\n"
            f"次に、ページ内の検索テキストボックスに『{query}』と直接入力し、同ページの検索ボタンをクリックして検索を実行してください。\n"
            f"要件:\n"
            f"- 最初の遷移のみ、ブラウザツールの検索/URLバーを使用して {starting_page} に移動する\n"
            f"- 以降は外部検索エンジン（Yahoo! JAPAN等）やアドレスバー検索は使わない\n"
            f"- ページの検索ボックスと検索ボタンのみを使用\n"
            f"- ページの読み込み完了を待ってから操作\n"
            f"- 検索結果の上位を確認し、日本語で特徴を3点に要約\n"
        )

        browser_use_agent = BrowserUseAgent(
            task=task,
            llm=bedrock_chat,
            browser_session=bu_session,
        )

        logger.info("running Browser-Use task: %s", task[:100] + "...")

        result = await browser_use_agent.run()

        return result
    finally:
        if bu_session:
            with contextlib.suppress(Exception):
                await bu_session.close()
                logger.info("Browser session closed")
        with contextlib.suppress(Exception):
            client.stop()
            logger.info("Browser client stopped")

@app.entrypoint
async def browser_agent(payload: Dict[str, Any], context) -> Dict[str, Any]:
    """
    Bedrock AgentCore Runtimeのエントリポイント
    contextパラメータを受け取り、適切な形式でレスポンスを返す
    """
    user_input = payload.get("prompt", "")

    model = BedrockModel(
    model_id="anthropic.claude-3-5-haiku-20241022-v1:0",
    params={"max_tokens": 2048, "temperature": 0.2},
    region="us-west-2",
    read_timeout=600,
    )

    agent = Agent(
        system_prompt=SYSTEM_PROMPT,
        model=model,
        tools=[run_browser_task],
    )

    try:
        result = agent(user_input)
        print(f"エージェント応答: {result}")
        return result

    except Exception as e:
        logger.error("Error in browser_agent: %s", str(e), exc_info=True)
        # エラーでも適切な形式で返す
        return {
            "output": {
                "error": str(e),
                "instruction": user_input,
                "message": f"エラーが発生しました: {str(e)}"
            }
        }

if __name__ == "__main__":
    app.run()

"""
デプロイ済みのBrowserエージェントを呼び出すテストスクリプト
"""
import boto3
import json
from botocore.config import Config

def invoke_browser_agent():
    agent_arn = "arn:aws:bedrock-agentcore:us-west-2:YOUR_ACCOUNT_ID:runtime/browser_agent-XXXXXXXXXX"

    if "YOUR_ACCOUNT_ID" in agent_arn:
        print("エラー: agent_arnを実際の値に置き換えてください")
        return None

    config = Config(
        read_timeout=600,
        connect_timeout=300,
        retries={'max_attempts': 2}
    )

    client = boto3.client("bedrock-agentcore", region_name="us-west-2", config=config)

    queries = [
        "ラ・ムー と検索して特徴を教えてください",
    ]

    for i, query in enumerate(queries, 1):
        print(f"\n{'='*60}")
        print(f"テスト {i}: {query}")
        print(f"{'='*60}")

        payload = json.dumps({"prompt": query}).encode("utf-8")

        try:
            response = client.invoke_agent_runtime(
                agentRuntimeArn=agent_arn,
                qualifier="DEFAULT",
                payload=payload,
                contentType="application/json",
                accept="application/json",
            )

            if response.get("contentType") == "application/json":
                content = []
                for chunk in response.get("response", []):
                    content.append(chunk.decode("utf-8"))

                try:
                    result = json.loads("".join(content))
                    print("エージェント応答:")
                    print(result)
                except json.JSONDecodeError:
                    print("Raw応答:")
                    raw_content = "".join(content)
                    print(raw_content)
            else:
                print(f"予期しないContent-Type: {response.get('contentType')}")
                print(f"レスポンス: {response}")
        except Exception as e:
            print(f"エラー: {e}")
            if hasattr(e, "response"):
                error_message = e.response.get("Error", {}).get("Message", "No message")
                print(f"エラーメッセージ: {error_message}")

if __name__ == "__main__":
    invoke_browser_agent()