PythonでAWS CDK経由のBedrock AgentをBright DataのSERP APIと併用

この記事では以下の内容を確認できます：

• AWS Cloud Development Kit (CDK) の概要と、クラウドインフラストラクチャの定義・デプロイへの活用方法
• – AWS CDKで構築したAWS Bedrock AIエージェントに、Bright DataのSERP APIのようなAI対応ツールを通じてウェブ検索結果へのアクセス権を与えるべき理由
• PythonでAWS CDKを使用し、SERP APIと統合されたAWS Bedrockエージェントを構築する方法。

それでは始めましょう！

AWS Cloud Development Kit（CDK）とは？

AWS Cloud Development Kit（AWS CDK）は、現代的なプログラミング言語を使用してクラウドインフラストラクチャをコードとして構築するためのオープンソースフレームワークです。TypeScript、Python、Java、C#、Goなどのプログラミング言語を使用して、AWSリソースをプロビジョニングし、AWS CloudFormation経由でアプリケーションをデプロイするために必要な機能を提供します。

AWS CDKを利用すれば、Amazon Bedrock向けAIエージェントもプログラムで構築できます。まさにこのチュートリアルで実現する内容です！

AWS CDKで構築したAmazon Bedrock AIエージェントがウェブ検索を必要とする理由

大規模言語モデルは、特定の時点までの知識しか反映していないデータセットで訓練されます。その結果、不正確な回答や幻覚的な回答を生成しがちです。これは最新情報を必要とするAIエージェントにとって特に問題となります。

この問題は、RAG（Retrieval-Augmented Generation）構成でAIエージェントに最新かつ信頼性の高いデータを取得させる機能を与えることで解決できます。例えば、AIエージェントがウェブ検索を実行して検証可能な情報を収集し、知識を拡張して精度を向上させることが可能です。

検索エンジンをスクレイピングするカスタムAWS Lambda関数の構築は可能ですが、非常に困難です。JavaScriptレンダリング、CAPTCHA、サイト構造の変更、IPブロックへの対応が必要となります。

より優れたアプローチは、Bright DataのSERP APIのような機能豊富なSERP APIを利用することです。これによりプロキシ処理、ブロック解除、スケーラビリティ、データフォーマットなどが自動的に管理されます。Lambda関数を用いてAWS Bedrockと統合することで、AWS CDKで構築したAIエージェントはリアルタイム検索結果にアクセスし、より信頼性の高い応答が可能になります。

PythonでAWS CDKとSERP API統合によるAIエージェント開発方法

このステップバイステップセクションでは、PythonとAWS CDKを使用してAWS Bedrock AIエージェントを構築する方法を学びます。このエージェントはBright Data SERP APIを介して検索エンジンからデータを取得できます。

統合は、エージェントがツールとして呼び出せるLambda関数（SERP APIを呼び出す）を通じて実装されます。具体的には、Amazon Bedrockエージェントを作成するための主要コンポーネントは以下の通りです：

• アクショングループ: エージェントが認識・呼び出し可能な関数を定義します。
• Lambda関数：Bright Data SERP APIをクエリするロジックを実装します。
• AIエージェント：基盤モデル、関数、ユーザーリクエスト間の相互作用を調整します。

この設定はPythonのAWS CDKで完全に実装されます。ビジュアルなAWS Bedrockコンソールで同様の結果を得るには、Amazon Bedrock + Bright Dataガイドを参照してください。

以下の手順に従い、AWS CDKを使用したAWS Bedrock AIエージェントを構築し、SERP APIを介したリアルタイムWeb検索機能で強化しましょう！

前提条件

このチュートリアルを実行するには、以下が必要です：

• AWS CDK CLIを使用するためのNode.js 22.x+のローカルインストール。
• Python 3.11以上がローカルにインストールされていること（PythonでAWS CDKを使用するため）。
• 有効なAWS アカウント（無料トライアルでも可）。
• Amazon Bedrock Agents の前提条件が設定されていること（Amazon Bedrock Agents は現在、 一部の AWS リージョンでのみ利用可能です）。
• APIキーが準備されたBright Dataアカウント。
• 基本的なPythonプログラミングスキル。

ステップ #1: AWS CLI のインストールと認証

AWS CDKを進める前に、AWS CLIをインストールし、ターミナルがAWSアカウントと認証できるように設定する必要があります。

注: AWS CLIが既にインストールされ、認証設定が完了している場合は、この手順をスキップして次の手順に進んでください。

お使いのオペレーティングシステムに対応した公式インストールガイドに従ってAWS CLIをインストールしてください。インストール後、以下のコマンドを実行してインストールを確認します：

aws --version

以下のような出力が表示されるはずです:

aws-cli/2.31.32 Python/3.13.9 Windows/11 exe/AMD64

次に、資格情報を設定するためにconfigureコマンドを実行します：

aws configure

入力が求められます:

1. AWS アクセスキー ID
2. AWS シークレットアクセスキー
3. デフォルトのリージョン名（例:us-east-1）
4. デフォルト出力形式（オプション、例:json）

CDKの開発とデプロイに必要なため、最初の3つのフィールドは必ず入力してください。これらの情報の入手方法がわからない場合は：

1. AWSにアクセスしてサインインします。
2. 右上のアカウント名をクリックしてアカウントメニューを開き、「セキュリティ認証情報」を選択します。
3. 「アクセスキー」セクションで、新しいキーを作成します。「アクセスキーID」と「シークレットアクセスキー」の両方を安全な場所に保存してください。

完了！これでマシンがCLI経由でAWSアカウントに接続可能になります。AWS CDK開発を進める準備が整いました。

ステップ #2: AWS CDK のインストール

aws-cdknpmパッケージを使用して、システムにAWS CDKをグローバルにインストールします：

npm install -g aws-cdk

次に、インストールされたバージョンを以下で確認します：

cdk --version

以下のような出力が表示されるはずです:

2.1031.2 (build 779352d)

注: Pythonを使用したAWS CDKによるAIエージェントの開発およびデプロイには、バージョン2.174.3以降が必要です。

完了！これでAWS CDK CLIがローカルにインストールされました。

ステップ #3: AWS CDK Python プロジェクトの設定

まず、AWS CDK + Bright Data SERP API AI エージェント用の新しいプロジェクトフォルダを作成します。
たとえば、次のように命名できます:

mkdir aws-cdk-bright-data-web-search-agent

フォルダに移動します：

cd aws-cdk-bright-data-web-search-agent

次に、initコマンドでPythonベースの新しいAWS CDKアプリケーションを初期化します：

cdk init app --language python

CDK CLIがプロジェクト構造を設定するまで時間がかかる場合がありますので、しばらくお待ちください。

初期化が完了すると、プロジェクトフォルダは次のようになります:

aws-cdk-bright-data-web-search-agent
├── .git/
├── venv/
├── aws_cdk_bright_data_web_search_agent/
│   ├── __init__.py
│   └── aws_cdk_bright_data_web_search_agent_stack.py
├── tests/
│   ├── __init__.py
│   └── unit/
│       ├── __init__.py
│       └── test_aws_cdk_bright_data_web_search_agent_stack.py
├── .gitignore
├── app.py
├── cdk.json
├── README.md
├── requirements.txt
├── requirements-dev.txt
└── source.bat

注目すべきは次の2つのファイルです:

• app.py: AWS CDK アプリケーションのトップレベル定義を含みます。
• aws_cdk_bright_data_web_search_agent/aws_cdk_bright_data_web_search_agent_stack.py: Web検索エージェント用のスタックを定義します（AIエージェントのロジックを実装する場所です）。

詳細は、PythonでのCDK操作に関する公式AWSガイドを参照してください。

次に、PyCharm やPython拡張機能を搭載したVisual Studio Codeなど、お好みのPython IDEでプロジェクトを読み込みます。

cdk initコマンドはプロジェクト内にPython 仮想環境を自動的に作成します。Linux または macOS では、以下で有効化してください：

source .venv/bin/activate

Windowsでは同等の操作として以下を実行：

.venvScriptsactivate

次に、アクティブ化した仮想環境内で必要な依存関係をすべてインストールします：

python -m pip install -r requirements.txt

素晴らしい！これでAIエージェント開発用のAWS CDK Python環境が設定されました。

ステップ #4: AWS CDK ブートストラップの実行

ブートストラップとは、AWS Cloud Development Kit（CDK）の使用に備えてAWS環境を準備するプロセスです。CDKスタックをデプロイする前に、環境のブートストラップが必須です。

簡単に言えば、このプロセスではAWSアカウントに以下のリソースを設定します：

• Amazon S3 バケット: AWS Lambda 関数コードやその他のアセットなど、CDK プロジェクトファイルを保存します。
• Amazon ECRリポジトリ: Dockerイメージを保存します。
• AWS IAMロール： AWS CDKがデプロイを実行するために必要な権限を付与します（詳細は、ブートストラップ中に作成されるIAMロールに関するAWSドキュメントを参照）。

CDKブートストラッププロセスを開始するには、プロジェクトフォルダで次のコマンドを実行します：

cdk bootstrap

このコマンドは、AWS CloudFormationサービス内で「CDKToolkit」というスタックを作成します。このスタックには、CDK アプリケーションのデプロイに必要なすべてのリソースが含まれます。

CloudFormationコンソールにアクセスし、「スタック」ページで確認してください：

「CDKToolkit」スタックが表示されます。そのリンクをクリックすると、以下のような内容が表示されます：

ブートストラッププロセスの動作、必要性、および内部処理の詳細については、公式の AWS CDK ドキュメントを参照してください。

ステップ #5: Bright Data の SERP API 準備

AWS CDK環境の開発・デプロイ準備が整ったところで、Bright Dataアカウントの準備とSERP APIサービスの設定という事前準備を完了させます。公式Bright Dataドキュメントに従うか、以下の手順を実行してください。

アカウントをお持ちでない場合は、Bright Dataアカウントを作成してください。または、ログインするだけで結構です。Bright Dataアカウントで、「Proxy & Scraping」ページにアクセスします。「My Zones」セクションで、テーブル内に「SERP API」行があるか確認してください：

「SERP API」ラベルの行が表示されない場合、ゾーンがまだ設定されていないことを意味します。「SERP API」セクションまでスクロールし、「ゾーンを作成」をクリックして追加します：

SERP APIゾーンを作成し、serp_api（または任意の名前）などの名前を付けます。API経由でサービスにアクセスする際に必要となるため、選択したゾーン名を覚えておいてください。

SERP API製品ページで「有効化」スイッチを切り替えてゾーンを有効化してください：

最後に、公式ガイドに従ってBright Data APIキーを生成してください。すぐに必要となるため、安全な場所に保管してください。

これで、AWS CDKで開発したAWS Bedrock AIエージェントでBright DataのSERP APIを使用するための準備が整いました。

ステップ #6: CDKアプリケーションのシークレットをAWS Secrets Managerに保存

機密情報（例：Bright Data APIキーやSERP APIゾーン名）を取得しました。これらの値をLambda関数のコードにハードコーディングする代わりに、AWS Secrets Managerから安全に読み取る必要があります。

以下のBashコマンドを実行し、Bright Data APIキーとSERP APIゾーンを含む「BRIGHT_DATA」という名前のシークレットを作成します：

aws secretsmanager create-secret 
  --name "BRIGHT_DATA" 
  --description "API credentials for Bright Data SERP API integration" 
  --secret-string '{
    "BRIGHT_DATA_API_KEY": "<YOUR_BRIGHT_DATA_API_KEY>",
    "BRIGHT_DATA_SERP_API_ZONE": "<YOUR_BRIGHT_DATA_SERP_API_ZONE>"
  }'

または、PowerShell では同等な方法で:

aws secretsmanager create-secret `
  --name "BRIGHT_DATA" `
  --description "Bright Data SERP API 統合用 API 認証情報" `
  --secret-string '{"BRIGHT_DATA_API_KEY":"<YOUR_BRIGHT_DATA_API_KEY>","BRIGHT_DATA_SERP_API_ZONE":"<YOUR_BRIGHT_DATA_SERP_API_ZONE>"}'

<YOUR_BRIGHT_DATA_API_KEY>および<YOUR_BRIGHT_DATA_SERP_API_ZONE>は、事前に取得した実際の値に置き換えてください。

このコマンドによりBRIGHT_DATAシークレットが設定されます。AWS Secrets Managerコンソールの「シークレット」ページで確認できます：

「シークレット値を取得」ボタンをクリックすると、BRIGHT_DATA_API_KEYおよびBRIGHT_DATA_SERP_API_ZONEシークレットが表示されます：

素晴らしい！これらのシークレットは、まもなく定義するLambda関数内でSERP APIへのリクエスト認証に使用されます。

ステップ #7: AWS CDK スタックの実装

AIエージェント構築に必要なすべての準備が整ったので、次のステップはPythonでAWS CDKスタックを実装することです。まず、AWS CDKスタックとは何かを理解することが重要です。

スタックはCDKにおける最小のデプロイ可能単位です。CDKコンストラクトを使用して定義されたAWSリソースの集合体を表します。CDKアプリケーションをデプロイすると、スタック内の全リソースが単一のCloudFormationスタックとして一括デプロイされます。

デフォルトのスタックファイルは次の場所にあります:

aws_cdk_bright_data_web_search_agent/aws_cdk_bright_data_web_search_agent_stack.py

Visual Studio Codeで確認します:

これは汎用スタックテンプレートを含み、ここにロジックを定義します。あなたのタスクは、Lambda関数、IAMロール、アクショングループ、Bedrock AIエージェントを含む、Bright Data SERP APIを統合したAIエージェントを構築するための完全なAWS CDKスタックを実装することです。

以下の方法で全てを実現してください：

import aws_cdk.aws_iam as iam
from aws_cdk import (
  Aws,
  CfnOutput,
  Duration,
  Stack)

from aws_cdk import aws_bedrock as bedrock
from aws_cdk import aws_lambda as _lambda
from constructs import Construct

# 必要な定数を定義
AI_MODEL_ID = "amazon.nova-lite-v1:0" # エージェントを駆動するLLMの名前
ACTION_GROUP_NAME = "action_group_web_search"
LAMBDA_FUNCTION_NAME = "serp_api_lambda"
AGENT_NAME = "web_search_agent"

# Bright Data搭載Web検索エージェント展開用CDKスタック定義
class AwsCdkBrightDataWebSearchAgentStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # ログ記録とシークレット読み取りのためのLambda権限を付与
        lambda_policy = iam.Policy(
            self,
            "LambdaPolicy",
            statements=[
                # CloudWatchロググループの作成権限
                iam.PolicyStatement(
                    sid="CreateLogGroup",
                    effect=iam.Effect.ALLOW,
                    actions=["logs:CreateLogGroup"],
                    resources=[f"arn:aws:logs:{Aws.REGION}:{Aws.ACCOUNT_ID}:*"],
                ),
                # ログストリームの作成とログイベントの書き込み権限
                iam.PolicyStatement(
                    sid="CreateLogStreamAndPutLogEvents",
                    effect=iam.Effect.ALLOW,
                    actions=["logs:CreateLogStream", "logs:PutLogEvents"],
                    resources=[
                        f"arn:aws:logs:{Aws.REGION}:{Aws.ACCOUNT_ID}:log-group:/aws/lambda/{LAMBDA_FUNCTION_NAME}",
                        f"arn:aws:logs:{Aws.REGION}:{Aws.ACCOUNT_ID}:log-group:/aws/lambda/{LAMBDA_FUNCTION_NAME}:log-stream:*",
                    ],
                ),
                # Secrets Manager から BRIGHT_DATA シークレットを読み取る権限
                iam.PolicyStatement(
                    sid="GetSecretsManagerSecret",
                    effect=iam.Effect.ALLOW,
                    actions=["secretsmanager:GetSecretValue"],
                    resources=[
                        f"arn:aws:secretsmanager:{Aws.REGION}:{Aws.ACCOUNT_ID}:secret:BRIGHT_DATA*",
                    ],
                ),
            ],
        )

        # Lambda関数用のIAMロールを定義
        lambda_role = iam.Role(
            self,
            "LambdaRole",
            role_name=f"{LAMBDA_FUNCTION_NAME}_role",
            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
        )

        # Lambdaロールにポリシーをアタッチ
        lambda_role.attach_inline_policy(lambda_policy)

        # Lambda関数の定義
        lambda_function = _lambda.Function(
            self,
            LAMBDA_FUNCTION_NAME,
            function_name=LAMBDA_FUNCTION_NAME,
            runtime=_lambda.Runtime.PYTHON_3_12,  # Pythonランタイム
            architecture=_lambda.Architecture.ARM_64,
            code=_lambda.Code.from_asset("lambda"),  # "lambda/"フォルダからLambdaコードを読み込む
            handler=f"{LAMBDA_FUNCTION_NAME}.lambda_handler",
            timeout=Duration.seconds(120),
            role=lambda_role,  # IAMロールをアタッチ
            environment={"LOG_LEVEL": "DEBUG", "ACTION_GROUP": f"{ACTION_GROUP_NAME}"},
        )

        # BedrockサービスがLambda関数を呼び出せるように許可
        bedrock_account_principal = iam.PrincipalWithConditions(
            iam.ServicePrincipal("bedrock.amazonaws.com"),
            conditions={
                "StringEquals": {"aws:SourceAccount": f"{Aws.ACCOUNT_ID}"},
            },
        )
        lambda_function.add_permission(
            id="LambdaResourcePolicyAgentsInvokeFunction",
            principal=bedrock_account_principal,
            action="lambda:invokeFunction",
        )

        # Bedrock エージェント用の IAM ポリシーを定義
        agent_policy = iam.Policy(
            self,
            "AgentPolicy",
            statements=[
                iam.PolicyStatement(
                    sid="AmazonBedrockAgentBedrockFoundationModelPolicy",
                    effect=iam.Effect.ALLOW,
                    actions=["bedrock:InvokeModel"],  # 基盤モデルの呼び出し権限を付与
                    resources=[f"arn:aws:bedrock:{Aws.REGION}::foundation-model/{AI_MODEL_ID}"],
                ),
            ],
        )

        # エージェントロールに対する信頼関係（Bedrockがロールをアサインできるようにするため）
        agent_role_trust = iam.PrincipalWithConditions(
            iam.ServicePrincipal("bedrock.amazonaws.com"),
            conditions={
                "StringLike": {"aws:SourceAccount": f"{Aws.ACCOUNT_ID}"},
                "ArnLike": {"aws:SourceArn": f"arn:aws:bedrock:{Aws.REGION}:{Aws.ACCOUNT_ID}:agent/*"},
            },
        )

        # Bedrockエージェント用のIAMロールを定義
        agent_role = iam.Role(
            self,
            "AmazonBedrockExecutionRoleForAgents",
            role_name=f"AmazonBedrockExecutionRoleForAgents_{AGENT_NAME}",
            assumed_by=agent_role_trust,
        )
        agent_role.attach_inline_policy(agent_policy)

        # AIエージェントのアクショングループを定義
        action_group = bedrock.CfnAgent.AgentActionGroupProperty(
            action_group_name=ACTION_GROUP_NAME,
            description="Bright DataのSERP APIを呼び出し、ウェブ検索を実行し、検索エンジンから最新の情報を取得します。",
            action_group_executor=bedrock.CfnAgent.ActionGroupExecutorProperty(lambda_=lambda_function.function_arn),
            function_schema=bedrock.CfnAgent.FunctionSchemaProperty(
                functions=[
                    bedrock.CfnAgent.FunctionProperty(
                        name=LAMBDA_FUNCTION_NAME,
                        description="Bright DataのSERP APIと連携し、ウェブ検索を実行します。",
                        parameters={
                            "search_query": bedrock.CfnAgent.ParameterDetailProperty(
                                type="string",
                                description="Googleウェブ検索の検索クエリ。",
                                required=True,
                            )
                        },
                    ),
                ]
            ),
        )

        # Bedrock AIエージェントの作成と指定
        agent_description = """
            Bright DataのSERP APIに接続し、検索エンジンから最新のウェブ検索コンテキストを取得できるAIエージェント。
        """

        agent_instruction = """
            あなたは、最新情報の取得と処理を必要とするユースケースを扱うために設計されたエージェントです。
            Bright DataのSERP APIを活用したウェブ検索を通じて、ニュースや検索エンジン結果を含む最新データにアクセスできます。
        """

        agent = bedrock.CfnAgent(
            self,
            AGENT_NAME,
            description=agent_description,
            agent_name=AGENT_NAME,
            foundation_model=AI_MODEL_ID,
            action_groups=[action_group],
            auto_prepare=True,
            instruction=agent_instruction,
            agent_resource_role_arn=agent_role.role_arn,
        )

        # 展開用の主要出力をエクスポート
        CfnOutput(self, "agent_id", value=agent.attr_agent_id)
        CfnOutput(self, "agent_version", value=agent.attr_agent_version)

上記のコードは、ウェブ検索AIエージェント向けの公式AWS CDKスタック実装を修正したものです。

この150行以上のコードで実行される処理は、ガイドで先に説明した手順（記事「Bright Data SERP API を AWS Bedrock AI エージェントに統合する方法」のステップ #1、ステップ #2、ステップ #3）を反映しています。

このファイルの動作をより深く理解するために、コードを5つの機能ブロックに分解しましょう：

1. Lambda IAMポリシーとロール：Lambda関数の権限を設定します。Lambdaは、実行詳細を記録するためにCloudWatchログへのアクセスと、Bright Data APIキーとゾーンを安全に読み取るためにAWS Secrets Managerへのアクセスを必要とします。Lambda用にIAMロールが作成され、適切なポリシーがアタッチされ、必要な権限のみを持って安全に動作することが保証されます。
2. Lambda関数定義：ここでLambda関数自体を定義・設定します。Pythonランタイムとアーキテクチャを指定し、Lambdaコードを含むフォルダ（次ステップで実装）を指し示し、ログレベルやアクショングループ名などの環境変数を設定します。AWS BedrockがLambdaを呼び出せる権限を付与し、AIエージェントがBright DataのSERP APIを通じてウェブ検索をトリガーできるようにします。
3. BedrockエージェントIAMロール: Bedrock AIエージェントの実行ロールを作成します。エージェントはサポート対象の 基盤AIモデル（本例ではamazon.nova-lite-v1:0）を呼び出す権限を必要とします。アカウント内でBedrockのみがロールを引き継げるよう信頼関係が定義され、モデルへのアクセス権を付与するポリシーがアタッチされます。
4. アクショングループ定義: AIエージェントが実行可能な具体的なアクションを定義します。エージェントをLambda関数にリンクし、Bright DataのSERP API経由でのウェブ検索を可能にします。また入力パラメータのメタデータを含み、エージェントが関数との連携方法や各検索に必要な情報を理解できるようにします。
5. Bedrock AIエージェント定義：AIエージェント自体を定義します。エージェントをアクショングループおよび実行ロールにリンクし、その使用に関する明確な説明と指示を提供します。

CDKスタックをデプロイすると、AWSコンソールにAIエージェントが表示されます。このエージェントはLambda関数内のBright Data SERP API統合を活用し、自律的にウェブ検索を実行して最新情報を取得できます。素晴らしい！

ステップ #8: SERP API 統合のための Lambda を実装する

以前のコードからこのスニペットをご覧ください：

lambda_function = _lambda.Function(
    self,
    LAMBDA_FUNCTION_NAME,
    function_name=LAMBDA_FUNCTION_NAME,
    runtime=_lambda.Runtime.PYTHON_3_12,  # Pythonランタイム
    architecture=_lambda.Architecture.ARM_64,
    code=_lambda.Code.from_asset("lambda"),  # "lambda/"フォルダからLambdaコードを読み込み
    handler=f"{LAMBDA_FUNCTION_NAME}.lambda_handler",
    timeout=Duration.seconds(120),
    role=lambda_role,  # IAMロールをアタッチ
    environment={"LOG_LEVEL": "DEBUG", "ACTION_GROUP": f"{ACTION_GROUP_NAME}"},
)

コード行 `code=_lambda.Code.from_asset("lambda")` は、Lambda関数コードをlambda/ フォルダから読み込むことを指定します。したがって、プロジェクト内にlambda/フォルダを作成し、その中にserp_api_lambda.pyという名前のファイルを追加してください:

serp_api_lambda.pyファイルには、先に定義したAIエージェントが使用するLambda関数の実装を含める必要があります。Bright Data SERP APIとの連携を処理するこの関数を以下のように実装してください:

import json
import logging
import os
import urllib.parse
import urllib.request

import boto3

# ----------------------------
# ログ設定
# ----------------------------
log_level = os.environ.get("LOG_LEVEL", "INFO").strip().upper()
logging.basicConfig(
    format="[%(asctime)s] p%(process)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
logger.setLevel(log_level)

# ----------------------------
# 環境変数からAWSリージョンを取得
# ----------------------------
AWS_REGION = os.environ.get("AWS_REGION")
if not AWS_REGION:
    logger.warning("AWS_REGION環境変数が設定されていません。boto3はデフォルトリージョンを使用します")

# ----------------------------
# AWS Secrets Managerからシークレットオブジェクトを取得
# ----------------------------
def get_secret_object(key: str) -> str:
    """
    AWS Secrets Manager からシークレット値を取得します。
    """
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=AWS_REGION
    )

    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=key
        )
    except Exception as e:
        logger.error(f"Secrets Managerからシークレット '{key}' を取得できませんでした: {e}")
        raise e

    secret = json.loads(get_secret_value_response["SecretString"])

    return secret


# Bright Data 認証情報の取得
bright_data_secret = get_secret_object("BRIGHT_DATA")
BRIGHT_DATA_API_KEY = bright_data_secret["BRIGHT_DATA_API_KEY"]
BRIGHT_DATA_SERP_API_ZONE = bright_data_secret["BRIGHT_DATA_SERP_API_ZONE"]

# ----------------------------
# SERP API Web検索
# ----------------------------
def serp_api_web_search(search_query: str) -> str:
    """
    Bright Data SERP APIを呼び出し、Google検索結果を取得します。
    """
    logger.info(f"search_query='{search_query}' に対する Bright Data SERP API 検索を実行中")

    # クエリをURL用にエンコード
    encoded_query = urllib.parse.quote(search_query)
    # SERPをスクレイピングするGoogle URLを構築
    search_engine_url = f"https://www.google.com/search?q={encoded_query}"

    # Bright Data APIリクエスト (ドキュメント: https://docs.brightdata.com/scraping-automation/serp-api/send-your-first-request)
    url = "https://api.brightdata.com/request"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {BRIGHT_DATA_API_KEY}"
    }
    data = {
        "zone": BRIGHT_DATA_SERP_API_ZONE,
        "url": search_engine_url,
        "format": "raw",
        "data_format": "markdown" # SERPをAI対応のMarkdown文書として取得するため
    }

    payload = json.dumps(data).encode("utf-8")
    request = urllib.request.Request(url, data=payload, headers=headers)

    try:
        response = urllib.request.urlopen(request)
        response_data = str(response.read().decode("utf-8"))
        logger.debug(f"SERP APIからの応答: {response_data}")
        return response_data
    except urllib.error.HTTPError as e:
        logger.error(f"Bright Data SERP APIの呼び出しに失敗しました。HTTPエラー {e.code}: {e.reason}")
    except urllib.error.URLError as e:
        logger.error(f"Bright Data SERP API の呼び出しに失敗しました。URL エラー: {e.reason}")

    return ""


# ----------------------------
# Lambda ハンドラー
# ----------------------------
def lambda_handler(event, _):
    """
    AWS Lambda ハンドラー。
    actionGroup、function、およびオプションのパラメータ（search_query を含む）を含むイベントを期待します。
    """
    logger.debug(f"lambda_handler がイベント: {event} で呼び出されました")

    action_group = event.get("actionGroup")
    function = event.get("function")
    parameters = event.get("parameters", [])

    # パラメータから検索クエリを抽出
    search_query = next(
        (param["value"] for param in parameters if param.get("name") == "search_query"),
        None,
    )
    logger.debug(f"入力検索クエリ: {search_query}")

    serp_page = serp_api_web_search(search_query) if search_query else ""
    logger.debug(f"検索クエリ結果: {serp_page}")

    # Lambdaレスポンスの準備
    function_response_body = {"TEXT": {"body": serp_page}}

    action_response = {
        "actionGroup": action_group,
        "function": function,
        "functionResponse": {"responseBody": function_response_body},
    }

    response = {"response": action_response, "messageVersion": event.get("messageVersion")}

    logger.debug(f"lambda_handler response: {response}")

    return response

このLambda関数は主に3つのタスクを処理します：

1. API認証情報を安全に取得: AWS Secrets ManagerからBright Data APIキーとSERP APIゾーンを取得し、機密情報がハードコードされないようにします。
2. SERP API経由でのウェブ検索の実行: 検索クエリをエンコードし、Google検索URLを構築して、Bright Data SERP APIにリクエストを送信します。APIは検索結果をMarkdown形式で返します。これはAIが消費するのに理想的なデータ形式です。
3. AWS Bedrockへの応答: AIエージェントが利用可能な構造化されたレスポンスとして結果を返します.

これで完了です！Bright Data SERP APIに接続するAWS Lambda関数が正常に定義されました。

ステップ #9: AWS CDK アプリケーションのデプロイ

CDKスタックと関連するLambda関数の実装が完了したので、最終ステップとしてスタックをAWSにデプロイします。プロジェクトフォルダ内で以下のデプロイコマンドを実行してください：

cdk deploy

IAMロール作成の許可を求められたら、yを入力して承認します。

しばらくすると、すべてが正常に動作していれば、次のような出力が表示されます：

次に、Amazon Bedrockコンソールに移動します。「エージェント」ページ内に「web_search_agent」エントリが表示されているはずです：

エージェントを開くと、デプロイされたエージェントの詳細が表示されます:

「編集とエージェントビルダー」ボタンを押して詳細を確認すると、「Bright Data SERP API を AWS Bedrock に統合する方法」で実装した AI エージェントと全く同じ構成が確認できます。

最後に、右隅のチャットインターフェースでエージェントを直接テストできる点に注意してください。これが次の最終ステップで行う作業です！

ステップ #10: AIエージェントのテスト

AIエージェントのウェブ検索とライブデータ取得機能をテストするには、次のようなプロンプトを試してください：

「米国政府閉鎖に関する最新ニュース記事を上位3件提示せよ」

（注：これは単なる例です。ウェブ検索結果を必要とする任意のプロンプトでテスト可能です）

このプロンプトは理想的です。なぜなら、基盤モデルだけでは知り得ない最新情報を要求しているためです。エージェントはSERP APIと連携したLambda関数を呼び出し、結果を取得し、データを処理して一貫性のある応答を生成します。

エージェントの「エージェントテスト」セクションでこのプロンプトを実行すると、以下のような出力が表示されるはずです：

内部では、エージェントがSERP API Lambda関数を呼び出し、「米国政府閉鎖」クエリに対する最新のGoogle検索結果を取得し、関連性の高い上位記事（URL付き）を抽出します。これは設定済みのNova Liteのような標準LLMでは単独では不可能な処理です。

詳細には、エージェントが生成した応答は以下の通りです：

選択されたニュース記事（およびそのURL）は、「米国政府閉鎖」クエリでGoogleを手動検索した場合の結果と一致します（少なくともエージェントテスト時点では）：

Google検索結果のスクレイピングを試みたことがある方なら、ボット検知、IP禁止、JavaScriptレンダリングなどの課題がどれほど困難かご存知でしょう。Bright Data SERP APIはこれらの問題をすべて処理し、AI最適化Markdown（またはHTML、JSONなど）形式でスクレイピングされたSERPを返します。

エージェントが実際にSERP API Lambda関数を呼び出したことを確認するには、応答ボックス内の「Show trace」ボタンをクリックします。「Trace step 1」セクションでグループ呼び出しログまでスクロールし、Lambda呼び出しの出力を確認してください：

これにより、Lambda関数が正常に実行され、エージェントが意図した通りにSERP APIとやり取りしたことを確認できます。Lambda関数のAWS CloudWatchログを確認して実行を検証することも可能です。

エージェントのさらなる進化へ！ファクトチェック、ブランド監視、市場トレンド分析など、様々なシナリオに関連するプロンプトをテストし、異なるエージェント機能やRAGユースケースにおけるパフォーマンスを確認しましょう。

これで完成です！PythonとAWS CDKライブラリを使用して、Bright DataのSERP APIと統合されたAWS Bedrockエージェントを構築しました。このAIエージェントは、要求に応じて最新で信頼性の高い文脈に沿ったウェブ検索データを取得できます。

まとめ

本ブログ記事では、AWS CDK Pythonプロジェクトを用いてBright DataのSERP APIをAWS Bedrock AIエージェントに統合する方法を解説しました。このワークフローは、AWS上でより文脈認識能力の高いAIエージェント構築を目指す開発者に最適です。

さらに高度なAIエージェントを作成するには、Bright DataのAI向けインフラストラクチャをご検討ください。ライブウェブデータの取得、検証、変換を行うツール群を提供しています。

今すぐBright Dataアカウントを無料で作成し、AI対応のウェブデータソリューションの実験を始めましょう！