技术分享 | 基于 Amazon Bedrock AgentCore + Web Search 构建 AI 竞争分析助手

import time
import boto3

gateway_client = boto3.client("bedrock-agentcore-control", region_name="us-east-1")
GATEWAY_ROLE_ARN = "arn:aws:iam::accountid:role/you-role-name"

#1. 创建 Gateway
gateway_response = gateway_client.create_gateway(
    name="competitive-intel-gateway",
    protocolType="MCP",
    description="Gateway for Competitive Intelligence Agent",
    roleArn=GATEWAY_ROLE_ARN,
    authorizerType="AWS_IAM"
)
gateway_id = gateway_response["gatewayId"]

#2. 轮询等待 Gateway 状态变为 ACTIVE
while True:
    status_response = gateway_client.get_gateway(gatewayIdentifier=gateway_id)
    current_status = status_response["status"]
    print(f"当前状态：{current_status}")

    if current_status == "READY":
        break
    elif current_status in ["FAILED", "DELETING"]:
        raise Exception(f"Gateway 创建失败，当前状态为：{current_status}")
    time.sleep(5)

#3. 添加 Web Search Target
print("正在关联 Web Search 工具...")
gateway_client.create_gateway_target(
    gatewayIdentifier=gateway_id,
    name="web-search-tool",
    targetConfiguration={
        "mcp": {
            "connector": {
                "source": {"connectorId": "web-search"},
                "configurations": [{"name": "WebSearch", "parameterValues": {}}],
            }
        }
    },
    credentialProviderConfigurations=[
        {"credentialProviderType": "GATEWAY_IAM_ROLE"}
    ],
)
print(f"Gateway {gateway_id} 已创建，Web Search 工具已成功启用！")

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "InvokeGateway",
            "Effect": "Allow",
            "Action": "bedrock-agentcore:InvokeGateway",
            "Resource": "arn:aws:bedrock-agentcore:us-east-1:account-id:gateway/competitive-intel-gateway-1s9kpehlxd"
        },
        {
            "Sid": "BedrockModelInvocation",
            "Effect": "Allow",
            "Action": [
                "bedrock:InvokeModel",
                "bedrock:InvokeModelWithResponseStream"
            ],
            "Resource": [
                "arn:aws:bedrock:*::foundation-model/*",
                "arn:aws:bedrock:us-east-1:account-id:*"
            ]
        },
        {
            "Sid": "InvokeWebSearch",
            "Effect": "Allow",
            "Action": "bedrock-agentcore:InvokeWebSearch",
            "Resource": "arn:aws:bedrock-agentcore:us-east-1:aws:tool/web-search.v1"
        }
    ]
}

"""
competitive_intel_agent.py
竞争分析 AI 助手 - 基于 Amazon Bedrock AgentCore + Web Search
"""
from datetime import date
from strands import Agent
from strands.models.bedrock import BedrockModel
from strands.tools.mcp import MCPClient
from mcp_proxy_for_aws.client import aws_iam_streamablehttp_client

#============================================================
#配置
#============================================================
GATEWAY_URL = "https://competitive-intel-gateway-1s9kpehlxd.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp" #替换成自己账号中的 url
AWS_REGION = "us-east-1"
MODEL_ID = "xxxxx"

#============================================================
#竞争分析专用 System Prompt
#============================================================

##你的角色
你为产品团队和战略团队提供实时、准确、有深度的竞品分析报告。你擅长从公开信息中洞察竞品的战略意图和市场动向
##分析方法论
当用户提出竞品分析需求时，严格按照以下流程执行：
###Phase 1: 需求理解
- 确认目标竞品名称
- 确认分析时间范围
- 确认关注的分析维度（产品/定价/市场/技术/生态）

###Phase 2: 多维度信息采集
对每个分析维度，制定 2-3 个精准的英文搜索查询（英文搜索覆盖面更广），从不同角度收集信息：
- **产品动态**: 新功能发布、产品路线图、版本更新
- **商业策略**: 定价变化、目标市场转移、营销活动
- **市场表现**: 分析师评价、市场份额、客户反馈
- **技术架构**: 技术博客、开源贡献、技术选型变化
- **生态合作**: 新合作伙伴、集成、收购

###Phase 3: 分析与洞察
- 识别关键事件并排列时间线
- 发现趋势和模式
- 评估对我方的潜在影响
- 提出应对建议

###Phase 4: 输出报告
严格按照以下格式输出：

---

#竞争情报报告：[竞品名称]

**分析时间范围**: [时间范围]
**报告生成时间**: {date.today().isoformat()}
**信息来源数**: [N] 条

#核心发现（Executive Summary）
> [3-5 句话概括最重要的发现]

##关键事件时间线
| 日期 | 事件 | 影响程度 | 来源 |
|------|------|---------|------|
| ... | ... | 高/中/低 | [链接] |

##产品动态分析
###新功能与更新
[细节分析]
###技术方向判断
[基于产品动态推断的技术路线]

##商业策略分析
###定价与包装变化
[如有变化，详细说明]
###目标市场与定位
[市场动向分析]

##生态与合作
[合作伙伴、集成、收购等动态]

##市场声音
###分析师观点
[来自 Gartner、Forrester 等的评价]
###客户与社区反馈
[开发者社区、社交媒体的讨论]

##对我方的影响评估
| 影响维度 | 影响程度 | 说明 |
|---------|---------|------|
| ... | 高/中/低 | ... |

##应对建议
1. [短期行动建议]
2. [中期策略调整]
3. [长期需要关注的方向]

##信息来源
[编号列出所有来源的标题、URL、发布时间]

##后续追踪建议
[建议持续关注的信号和触发条件]

---

##重要原则
- 每条重要信息必须标注来源 URL
- 区分"事实"和"推测"，推测部分需明确标注
- 如果某个维度没有搜索到有效信息，诚实说明
- 英文搜索优先（覆盖面更广），中文补充搜索
- 关注最近 3 个月的信息，优先呈现最新动态
"""

#============================================================
#Agent 创建与执行
#============================================================
def create_competitive_intel_agent():
    mcp_client = MCPClient(lambda: aws_iam_streamablehttp_client(
        endpoint=GATEWAY_URL,
        aws_region=AWS_REGION,
        aws_service="bedrock-agentcore",
    ))
    model = BedrockModel(model_id=MODEL_ID)
    return mcp_client, model
def analyze_competitor(query: str) -> str:
    """
    执行竞争分析

    Args:
        query: 用户的分析需求，例如 "分析 Datadog 近三个月的产品动态"

    Returns:
        结构化的竞争情报报告
    """
    mcp_client, model = create_competitive_intel_agent()
    with mcp_client:

        #自动发现 Gateway 上的 Web Search 工具
        tools = mcp_client.list_tools_sync()
        print(f"已连接 AgentCore Gateway, 发现 {len(tools)} 个工具")

        #创建 Agent
        agent = Agent(
            model=model,
            tools=tools,
            system_prompt=SYSTEM_PROMPT
        )

        #执行分析
        print(f"开始竞争分析：{query}")
        print("=" * 60)
        result = agent(query)
        print("=" * 60)
        print("分析完成")
        return result
class CompetitiveIntelTracker:
    """持续追踪模式 - 记住历史分析，识别变化趋势"""

    def __init__(self, memory_store_id: str):
        self.memory_store_id = memory_store_id
        self.mcp_client, self.model = create_competitive_intel_agent()

    def analyze(self, query: str, session_id: str = "default") -> str:
        """带上下文记忆的分析"""
        with self.mcp_client:
            tools = self.mcp_client.list_tools_sync()
            agent = Agent(
                model=self.model,
                tools=tools,
                system_prompt=SYSTEM_PROMPT + f"""

\n\n##历史追踪模式
你正在持续追踪模式下工作。利用 Memory 中的历史分析记录：
- 对比本次发现与上次分析的变化
- 标注"新发现"vs"持续趋势"
- 在报告末尾增加"与上次分析对比"章节
""",
                memory_config={
                    "memory_store_id": self.memory_store_id,
                    "session_id": session_id,
                    "namespace": "competitive-intel"
                }
            )
            return agent(query)

#创建 ECR 仓库
aws ecr create-repository --repository-name competitive-intel-agent --region us-east-1

#登录 ECR
aws ecr get-login-password --region us-east-1 | \
    docker login --username AWS --password-stdin <ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com

#构建镜像
docker build -t competitive-intel-agent .

#打 Tag 并推送
docker tag competitive-intel-agent:latest \
    <ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com/competitive-intel-agent:latest
docker push <ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com/competitive-intel-agent:latest

import boto3
runtime_client = boto3.client("bedrock-agentcore-control", region_name="us-east-1")
#部署 Agent
response = runtime_client.create_agent_runtime(
    agentRuntimeName="competitive-intel-agent",
    description="AI Competitive Intelligence Agent with real-time web search",
    agentRuntimeArtifact={
        "containerConfiguration": {
            "containerUri": "<ACCOUNT_ID>.dkr.ecr.us-east-1.amazonaws.com/competitive-intel-agent:latest"
        }
    },
    roleArn="arn:aws:iam::<ACCOUNT_ID>:role/AgentCoreRuntimeRole",
    networkConfiguration={"networkMode": "PUBLIC"}
)

#创建调用端点
endpoint = runtime_client.create_agent_runtime_endpoint(
    agentRuntimeId=response["agentRun-timeId"],
    name="competitive-intel-endpoint"
)
print(f" Agent 已部署：{endpoint['agentRuntimeEndpoint']['url']}")

关于伊克罗德信息