通过 AWS CDK 在 Python 中使用 Bright Data 的 搜索引擎 API 构建 Bedrock Agent

在本文中，你将看到：

• 什么是 AWS Cloud Development Kit（CDK），以及如何使用它来定义和部署云基础设施。
• 为什么你应该让使用 AWS CDK 构建的 AWS Bedrock AI agent，通过像 Bright Data 的 搜索引擎 API 这样的 AI 就绪工具访问网页搜索结果。
• 如何在 Python 中使用 AWS CDK 构建一个与 搜索引擎 API 集成的 AWS Bedrock agent。

让我们开始吧！

什么是 AWS Cloud Development Kit（CDK）？

AWS Cloud Development Kit，也称为 AWS CDK，是一个开源框架，用于使用现代编程语言以基础设施即代码的方式构建云基础设施。它为你提供了通过 AWS CloudFormation 使用 TypeScript、Python、Java、C# 和 Go 等编程语言来预置 AWS 资源并部署应用程序所需的一切。

得益于 AWS CDK，你还可以以编程方式构建 Amazon Bedrock 的 AI agent——这正是你将在本教程中要做的！

为什么使用 AWS CDK 构建的 Amazon Bedrock AI agent 需要网页搜索

大型语言模型是基于数据集训练的，这些数据集只代表截至某个时间点的知识。因此，它们往往会产生不准确或幻觉式的回答。这对于需要最新信息的 AI agent 来说尤其成问题。

这可以通过在 RAG（检索增强生成）设置中赋予你的 AI agent 获取新鲜、可靠数据的能力来解决。例如，AI agent 可以执行网页搜索来收集可验证的信息，扩展其知识并提高准确性。

构建一个自定义的 AWS Lambda 函数来抓取搜索引擎是可行的，但相当具有挑战性。你必须处理 JavaScript 渲染、验证码破解、不断变化的网站结构以及 IP 封锁。

更好的方法是使用功能丰富的 搜索引擎 API，例如 Bright Data 的 搜索引擎 API。它会为你处理代理、解锁、可扩展性、数据格式化等更多内容。通过使用 Lambda 函数将其与 AWS Bedrock 集成，你通过 AWS CDK 构建的 AI agent 将能够访问实时搜索结果，从而获得更可信的响应。

如何在 Python 中使用 AWS CDK 开发集成 搜索引擎 API 的 AI Agent

在本分步章节中，你将学习如何在 Python 中使用 AWS CDK 构建一个 AWS Bedrock AI agent。它将能够通过 Bright Data 搜索引擎 API 从搜索引擎获取数据。

该集成将通过一个 Lambda 函数（调用 搜索引擎 API）来实现，agent 可以将其作为工具调用。具体来说，要创建一个 Amazon Bedrock agent，主要组件包括：

• Action Group：定义 agent 可以看到并调用的函数。
• Lambda Function：实现查询 Bright Data 搜索引擎 API 的逻辑。
• AI Agent：编排基础模型、函数与用户请求之间的交互。

此设置将完全使用 Python 中的 AWS CDK 来实现。要使用可视化的 AWS Bedrock 控制台实现相同结果，请参阅我们的 Amazon Bedrock + Bright Data 指南。

按照以下步骤使用 AWS CDK 构建一个 AWS Bedrock AI agent，并通过 搜索引擎 API 增强实时网页搜索能力！

先决条件

要跟随本教程，你需要：

• 本地安装 Node.js 22.x+ 以使用 AWS CDK CLI。
• 本地安装 Python 3.11+ 以使用 Python 中的 AWS CDK。
• 一个有效的 AWS 账户（即使是免费试用也可以）。
• 已设置 Amazon Bedrock Agents 的先决条件。（Amazon Bedrock Agents 目前 仅在部分 AWS 区域可用。）
• 一个已准备好 API key 的 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 Access Key ID
2. AWS Secret Access Key
3. 默认区域名称（例如 us-east-1）
4. 默认输出格式（可选，例如 json）

填写前三个字段，因为它们是 CDK 开发和部署所必需的。如果你想知道在哪里获取这些信息：

1. 前往 AWS 并登录。
2. 在右上角，点击你的账户名称以打开账户菜单，并选择“Security Credentials”选项。
3. 在“Access Keys”部分下，创建一个新 key。将“Access Key ID”和“Secret Access Key”都保存在安全的地方。

完成！你的机器可以通过 CLI 连接到你的 AWS 账户。你已准备好继续进行 AWS CDK 开发。

步骤 #2：安装 AWS CDK

使用 aws-cdk npm 包，在你的系统上全局安装 AWS CDK：

npm install -g aws-cdk

然后，通过运行以下命令验证已安装的版本：

cdk --version

你应该会看到类似于以下的输出：

2.1031.2 (build 779352d)

注意：使用 Python 的 AWS CDK 进行 AI agent 开发和部署需要 2.174.3 或更高版本。

很好！你现在已在本地安装了 AWS CDK CLI。

步骤 #3：设置你的 AWS CDK Python 项目

首先，为你的 AWS CDK + Bright Data 搜索引擎 API AI agent 创建一个新的项目文件夹。
例如，你可以将其命名为 aws-cdk-bright-data-web-search-agent：

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

你需要重点关注的是这两个文件：

• app.py：包含 AWS CDK 应用程序的顶层定义。
• aws_cdk_bright_data_web_search_agent/aws_cdk_bright_data_web_search_agent_stack.py：定义网页搜索 agent 的 stack（你将在这里实现你的 AI agent 逻辑）。

有关更多详细信息，请参阅 AWS 官方指南：在 Python 中使用 CDK。

现在，在你喜欢的 Python IDE 中加载你的项目，例如 PyCharm 或 带有 Python 扩展的 Visual Studio Code。

注意，cdk init 命令会在项目中自动创建一个 Python 虚拟环境。在 Linux 或 macOS 上，使用以下命令激活它：

source .venv/bin/activate

或者在 Windows 上运行：

.venvScriptsactivate

然后，在已激活的虚拟环境中，安装所有必需的依赖项：

python -m pip install -r requirements.txt

太棒了！你现在已经为 AI agent 开发设置好了 AWS CDK Python 环境。

步骤 #4：运行 AWS CDK Bootstrapping

Bootstrapping 是为 AWS Cloud Development Kit 的使用准备你的 AWS 环境的过程。在部署 CDK stack 之前，你的环境必须完成 bootstrapped。

更简单地说，此过程会在你的 AWS 账户中设置以下资源：

• 一个 Amazon S3 bucket：存储你的 CDK 项目文件，例如 AWS Lambda 函数代码和其他资产。
• 一个 Amazon ECR repository：存储 Docker 镜像。
• AWS IAM roles：授予 AWS CDK 执行部署所需的权限。（有关更多详细信息，请参阅 AWS 文档：bootstrapping 期间创建的 IAM roles。）

要启动 CDK bootstrap 过程，请在你的项目文件夹中运行以下命令：

cdk bootstrap

在 AWS CloudFormation 服务中，该命令会创建一个名为“CDKToolkit”的 stack，其中包含部署 CDK 应用程序所需的所有资源。

通过进入 CloudFormation 控制台并检查“Stacks”页面来验证这一点：

你将看到一个“CDKToolkit”stack。点击其链接，你应该会看到类似如下内容：

有关 bootstrapping 过程如何工作、为什么需要它以及幕后发生了什么的更多信息，请参阅 AWS CDK 官方文档。

步骤 #5：准备 Bright Data 的 搜索引擎 API

现在你的 AWS CDK 环境已为开发和部署设置完成，通过准备你的 Bright Data 账户并配置 搜索引擎 API 服务来完成初步步骤。你可以遵循 Bright Data 官方文档，或按照以下步骤操作。

如果你还没有账户，请创建一个 Bright Data 账户。或者，直接登录。在你的 Bright Data 账户中，进入“Proxies & Scraping”页面。在“My Zones”部分，检查表格中是否有一行标注为“搜索引擎 API”：

如果你没有看到带有“搜索引擎 API”标签的行，这意味着你尚未设置一个 区域。向下滚动到“搜索引擎 API”部分并点击“Create 区域”以添加一个：

创建一个 搜索引擎 API 区域，并给它一个名称，例如 serp_api（或任何你喜欢的名称）。记住你选择的 区域 名称，因为你将需要它通过 API 访问该服务。

在 搜索引擎 API 产品页面上，切换“Activate”开关以启用该 区域：

最后，按照官方指南生成你的 Bright Data API key。将其存放在安全的地方，因为你很快就会用到它。

太棒了！你现在已经完成了所有设置，可以在使用 AWS CDK 开发的 AWS Bedrock AI agent 中使用 Bright Data 的 搜索引擎 API。

步骤 #6：将你的 CDK 应用程序密钥存储在 AWS Secrets Manager 中

你刚刚获得了敏感信息（例如你的 Bright Data API key 和 搜索引擎 API 区域 名称）。与其在 Lambda 函数代码中硬编码这些值，你应该从 AWS Secrets Manager 中安全地读取它们。

运行以下 Bash 命令创建一个名为 BRIGHT_DATA 的 secret，其中包含你的 Bright Data API key 和 搜索引擎 API 区域：

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 "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>"}'

请确保将 <YOUR_BRIGHT_DATA_API_KEY> 和 <YOUR_BRIGHT_DATA_SERP_API_ZONE> 替换为你之前检索到的实际值。

该命令将设置 BRIGHT_DATA secret，你可以在 AWS Secrets Manager 控制台的“Secrets”页面下确认：

如果你点击“Retrieve secret value”按钮，你应该会看到 BRIGHT_DATA_API_KEY 和 BRIGHT_DATA_SERP_API_ZONE secrets：

太棒了！这些 secrets 将用于在你即将定义的 Lambda 函数中对 搜索引擎 API 的请求进行身份验证。

步骤 #7：实现你的 AWS CDK Stack

现在你已经设置好了构建 AI agent 所需的一切，下一步是在 Python 中实现 AWS CDK stack。首先，理解什么是 AWS CDK stack 很重要。

stack 是 CDK 中最小的可部署单元。它表示使用 CDK constructs 定义的一组 AWS 资源。当你部署一个 CDK app 时，stack 中的所有资源会作为一个单一的 CloudFormation stack 一起部署。

默认的 stack 文件位于：

aws_cdk_bright_data_web_search_agent/aws_cdk_bright_data_web_search_agent_stack.py

在 Visual Studio Code 中检查它：

这包含一个通用的 stack 模板，你将在其中定义你的逻辑。你的任务是实现完整的 AWS CDK stack，以构建与 Bright Data 搜索引擎 API 集成的 AI agent，包括 Lambda functions、IAM roles、action groups 以及 Bedrock AI agent。

通过以下方式实现所有这些：

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

# Define the required constants
AI_MODEL_ID = "amazon.nova-lite-v1:0" # The name of the LLM used to power the agent
ACTION_GROUP_NAME = "action_group_web_search"
LAMBDA_FUNCTION_NAME = "serp_api_lambda"
AGENT_NAME = "web_search_agent"

# Define the CDK Stack for deploying the Bright Data-powered Web Search Agent
class AwsCdkBrightDataWebSearchAgentStack(Stack):

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

        # Grants Lambda permissions for logging and reading secrets
        lambda_policy = iam.Policy(
            self,
            "LambdaPolicy",
            statements=[
                # Permission to create CloudWatch log groups
                iam.PolicyStatement(
                    sid="CreateLogGroup",
                    effect=iam.Effect.ALLOW,
                    actions=["logs:CreateLogGroup"],
                    resources=[f"arn:aws:logs:{Aws.REGION}:{Aws.ACCOUNT_ID}:*"],
                ),
                # Permission to create log streams and put log events
                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:*",
                    ],
                ),
                # Permission to read BRIGHT_DATA secrets from Secrets Manager
                iam.PolicyStatement(
                    sid="GetSecretsManagerSecret",
                    effect=iam.Effect.ALLOW,
                    actions=["secretsmanager:GetSecretValue"],
                    resources=[
                        f"arn:aws:secretsmanager:{Aws.REGION}:{Aws.ACCOUNT_ID}:secret:BRIGHT_DATA*",
                    ],
                ),
            ],
        )

        # Define the IAM role for the Lambda functions
        lambda_role = iam.Role(
            self,
            "LambdaRole",
            role_name=f"{LAMBDA_FUNCTION_NAME}_role",
            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
        )

        # Attach the policy to the Lambda role
        lambda_role.attach_inline_policy(lambda_policy)

        # Lambda function definition
        lambda_function = _lambda.Function(
            self,
            LAMBDA_FUNCTION_NAME,
            function_name=LAMBDA_FUNCTION_NAME,
            runtime=_lambda.Runtime.PYTHON_3_12,  # Python runtime
            architecture=_lambda.Architecture.ARM_64,
            code=_lambda.Code.from_asset("lambda"),  # Read the Lambda code from the "lambda/" folder
            handler=f"{LAMBDA_FUNCTION_NAME}.lambda_handler",
            timeout=Duration.seconds(120),
            role=lambda_role,  # Attach IAM role
            environment={"LOG_LEVEL": "DEBUG", "ACTION_GROUP": f"{ACTION_GROUP_NAME}"},
        )

        # Allow the Bedrock service to invoke Lambda functions
        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",
        )

        # Define the IAM Policy for the Bedrock agent
        agent_policy = iam.Policy(
            self,
            "AgentPolicy",
            statements=[
                iam.PolicyStatement(
                    sid="AmazonBedrockAgentBedrockFoundationModelPolicy",
                    effect=iam.Effect.ALLOW,
                    actions=["bedrock:InvokeModel"],  # Give it the permission to call foundation model
                    resources=[f"arn:aws:bedrock:{Aws.REGION}::foundation-model/{AI_MODEL_ID}"],
                ),
            ],
        )

        # Trust relationship for agent role to allow Bedrock to assume it
        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/*"},
            },
        )

        # Define the IAM role for the Bedrock agent
        agent_role = iam.Role(
            self,
            "AmazonBedrockExecutionRoleForAgents",
            role_name=f"AmazonBedrockExecutionRoleForAgents_{AGENT_NAME}",
            assumed_by=agent_role_trust,
        )
        agent_role.attach_inline_policy(agent_policy)

        # Define the action group for the AI agent
        action_group = bedrock.CfnAgent.AgentActionGroupProperty(
            action_group_name=ACTION_GROUP_NAME,
            description="Call Bright Data's SERP API to perform web searches and retrieve up-to-date information from search engines.",
            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="Integrates with Bright Data's SERP API to perform web searches.",
                        parameters={
                            "search_query": bedrock.CfnAgent.ParameterDetailProperty(
                                type="string",
                                description="The search query for the Google web search.",
                                required=True,
                            )
                        },
                    ),
                ]
            ),
        )

        # Create and specify the Bedrock AI Agent
        agent_description = """
            An AI agent that can connect to Bright Data's SERP API to retrieve fresh web search context from search engines.
        """

        agent_instruction = """
            You are an agent designed to handle use cases that require retrieving and processing up-to-date information.
            You can access current data, including news and search engine results, through web searches powered by Bright Data's 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,
        )

        # Export the key outputs for deployment
        CfnOutput(self, "agent_id", value=agent.attr_agent_id)
        CfnOutput(self, "agent_version", value=agent.attr_agent_version)

上述代码是官方 用于网页搜索 AI agent 的 AWS CDK stack 实现 的修改版本。

这 150+ 行代码中发生的事情反映了本指南前面概述的步骤：我们文章“如何将 Bright Data 搜索引擎 API 与 AWS Bedrock AI Agents 集成”中的步骤 #1、步骤 #2 和步骤 #3。

为了更好地理解该文件中发生了什么，让我们将代码分解为五个功能块：

1. Lambda IAM policy and role：为 Lambda function 设置权限。Lambda 需要访问 CloudWatch logs 以记录执行细节，并访问 AWS Secrets Manager 以安全地读取 Bright Data API key 和 区域。为 Lambda 创建一个 IAM role，并附加相应的 policy，确保它以安全方式运行，并且只拥有它所需的权限。
2. Lambda function definition：在这里定义并配置 Lambda function 本身。它指定 Python runtime 和架构，指向包含 Lambda 代码的文件夹（将在下一步实现），并设置环境变量，例如日志级别和 action group 名称。授予权限以便 AWS Bedrock 可以调用 Lambda，从而允许 AI agent 通过 Bright Data 的 搜索引擎 API 触发网页搜索。
3. Bedrock agent IAM role：为 Bedrock AI agent 创建执行 role。agent 需要权限来调用所选的基础 AI model（来自支持的模型，此处为 amazon.nova-lite-v1:0）。定义信任关系，使得只有 Bedrock 可以在你的账户内承担该 role。附加一个 policy 以授予对该 model 的访问权限。
4. Action group definition：action group 定义 AI agent 可以执行的特定操作。它将 agent 链接到 Lambda function，使其能够通过 Bright Data 的 搜索引擎 API 执行网页搜索。action group 还包含输入参数的元数据，因此 agent 能理解如何与该 function 交互，以及每次搜索需要哪些信息。
5. Bedrock AI agent definition：定义 AI agent 本身。它将 agent 链接到 action group 和执行 role，并为其使用提供清晰的描述和说明。

在部署 CDK stack 后，一个 AI agent 将出现在你的 AWS 控制台中。该 agent 可以自主执行网页搜索，并通过利用 Lambda function 中的 Bright Data 搜索引擎 API 集成来检索最新信息。太棒了！

步骤 #8：实现用于 搜索引擎 API 集成的 Lambda

看看前一段代码中的这个片段：

lambda_function = _lambda.Function(
    self,
    LAMBDA_FUNCTION_NAME,
    function_name=LAMBDA_FUNCTION_NAME,
    runtime=_lambda.Runtime.PYTHON_3_12,  # Python runtime
    architecture=_lambda.Architecture.ARM_64,
    code=_lambda.Code.from_asset("lambda"),  # Read the Lambda code from the "lambda/" folder
    handler=f"{LAMBDA_FUNCTION_NAME}.lambda_handler",
    timeout=Duration.seconds(120),
    role=lambda_role,  # Attach IAM role
    environment={"LOG_LEVEL": "DEBUG", "ACTION_GROUP": f"{ACTION_GROUP_NAME}"},
)

code=_lambda.Code.from_asset("lambda") 这一行指定 Lambda function 代码将从 lambda/ 文件夹加载。因此，在你的项目中创建一个 lambda/ 文件夹，并在其中添加一个名为 serp_api_lambda.py 的文件：

serp_api_lambda.py 文件需要包含前面定义的 AI agent 所使用的 Lambda function 的实现。按如下方式实现该函数以处理与 Bright Data 搜索引擎 API 的集成：

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

import boto3

# ----------------------------
# Logging configuration
# ----------------------------
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 Region from environment
# ----------------------------
AWS_REGION = os.environ.get("AWS_REGION")
if not AWS_REGION:
    logger.warning("AWS_REGION environment variable not set; boto3 will use default region")

# ----------------------------
# Retrieve the secret object from AWS Secrets Manager
# ----------------------------
def get_secret_object(key: str) -> str:
    """
    Get a secret value from 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"Could not get secret '{key}' from Secrets Manager: {e}")
        raise e

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

    return secret


# Retrieve Bright Data credentials
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 Search
# ----------------------------
def serp_api_web_search(search_query: str) -> str:
    """
    Calls Bright Data SERP API to retrieve Google search results.
    """
    logger.info(f"Executing Bright Data SERP API search for search_query='{search_query}'")

    # Encode the query for URL
    encoded_query = urllib.parse.quote(search_query)
    # Build the Google URL to scrape the SERP from
    search_engine_url = f"https://www.google.com/search?q={encoded_query}"

    # Bright Data API request (docs: 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" # To get the SERP as an AI-ready Markdown document
    }

    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"Response from SERP API: {response_data}")
        return response_data
    except urllib.error.HTTPError as e:
        logger.error(f"Failed to call Bright Data SERP API. HTTP Error {e.code}: {e.reason}")
    except urllib.error.URLError as e:
        logger.error(f"Failed to call Bright Data SERP API. URL Error: {e.reason}")

    return ""


# ----------------------------
# Lambda handler
# ----------------------------
def lambda_handler(event, _):
    """
    AWS Lambda handler.
    Expects event with actionGroup, function, and optional parameters (including search_query).
    """
    logger.debug(f"lambda_handler called with event: {event}")

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

    # Extract search_query from parameters
    search_query = next(
        (param["value"] for param in parameters if param.get("name") == "search_query"),
        None,
    )
    logger.debug(f"Input search query: {search_query}")

    serp_page = serp_api_web_search(search_query) if search_query else ""
    logger.debug(f"Search query results: {serp_page}")

    # Prepare the Lambda response
    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 function 处理三个主要任务：

1. 安全地检索 API 凭证：从 AWS Secrets Manager 获取 Bright Data API key 和 搜索引擎 API 区域，因此敏感信息永远不会被硬编码。
2. 通过 搜索引擎 API 执行网页搜索：对搜索查询进行编码，构建 Google 搜索 URL，并向 Bright Data 搜索引擎 API 发送请求。该 API 返回以 Markdown 格式化的搜索结果，这是一种适合 AI 消费的理想数据格式。
3. 响应 AWS Bedrock：以 AI agent 可用的结构化响应返回结果。

开始吧！你用于连接 Bright Data 搜索引擎 API 的 AWS Lambda function 现在已经成功定义。

步骤 #9：部署你的 AWS CDK 应用程序

现在你的 CDK stack 及其相关的 Lambda function 已经实现，最后一步是将你的 stack 部署到 AWS。为此，在你的项目文件夹中运行 deploy 命令：

cdk deploy

当提示你授予创建 IAM role 的权限时，输入 y 以批准。

几分钟后，如果一切按预期工作，你应该会看到类似如下的输出：

接下来，前往 Amazon Bedrock 控制台。在“Agents”页面中，你应该会注意到一个 “web_search_agent” 条目：

打开该 agent，你将看到已部署的 agent 详情：

通过按下“Edit and Agent Builder”按钮来检查它，你将看到与“如何将 Bright Data 搜索引擎 API 与 AWS Bedrock 集成”中实现的 AI agent 完全相同。

最后，请注意你可以直接使用右侧角落的聊天界面测试该 agent。这就是你将在下一步也是最后一步要做的事情！

步骤 #10：测试 AI Agent

要测试你的 AI agent 的网页搜索与实时数据检索能力，尝试如下提示：

"Give me the top 3 latest news articles on the US government shutdown"

(注意：这只是一个示例。因此，你可以测试任何需要网页搜索结果的提示。)

这是一个理想的提示，因为它请求的是仅靠基础模型本身并不知道的最新信息。agent 将调用与 搜索引擎 API 集成的 Lambda function，检索结果，并处理数据以生成连贯的响应。

在你的 agent 的“Test Agent”部分运行该提示，你应该会看到类似如下的输出：

在幕后，agent 调用 搜索引擎 API Lambda function，检索“US government shutdown”查询的最新 Google 搜索结果，并提取最相关的顶部文章（以及它们的 URLs）。这是像已配置的 Nova Lite 这样的标准 LLM 无法自行完成的事情。

详细来说，这是 agent 生成的响应：

所选的新闻文章（及其 URLs）与您在 Google 上针对“US government shutdown”查询手动找到的内容一致（至少截至测试该 agent 的日期）：

现在，任何尝试过抓取 Google 搜索结果的人都知道，由于机器人检测、IP 封禁、JavaScript 渲染以及其他挑战，这会有多么困难。Bright Data 搜索引擎 API 为你处理所有这些问题，以 AI 优化的 Markdown（或 HTML、JSON 等）格式返回抓取的 SERPs。

要确认你的 agent 确实调用了 搜索引擎 API Lambda function，请点击响应框中的“Show trace”按钮。在“Trace step 1”部分，向下滚动到 group invocation log 以检查来自 Lambda 调用的输出：

这验证了 Lambda function 成功执行，并且 agent 按预期与 搜索引擎 API 进行了交互。你还可以检查你的 Lambda function 的 AWS CloudWatch logs 以确认执行情况。

是时候进一步推进你的 agent 了！测试与事实核查、品牌监控、市场趋势分析或其他场景相关的提示，看看它在不同的agentic 和 RAG 用例中表现如何。

Et voilà！你已成功使用 Python 和 AWS CDK 库构建了一个与 Bright Data 的 搜索引擎 API 集成的 AWS Bedrock agent。该 AI agent 能够按需检索最新、可靠且具上下文的网页搜索数据。

结论

在这篇博客文章中，你学习了如何在 AWS CDK Python 项目中将 Bright Data 的 搜索引擎 API 集成到 AWS Bedrock AI agent 中。该工作流非常适合希望在 AWS 上构建更具上下文感知能力的 AI agent 的开发者。

要创建更复杂的 AI agent，请探索 Bright Data 面向 AI 的基础设施。它提供了一套用于检索、验证和转换实时网页数据的工具。

立即免费创建一个 Bright Data 账户，并开始试验我们面向 AI 的网页数据解决方案！