bedrock-agentcore

Amazon Bedrock AgentCore

概览

Amazon Bedrock AgentCore 是一个用于安全、大规模构建、部署和运行高效 AI Agent 的智能体平台，无需管理基础设施。它提供与框架无关的基础能力，可与热门开源框架（Strands、LangGraph、CrewAI、Autogen）以及任意模型配合使用。

用途：借助企业级基础设施，将任意 AI Agent 转变为可用于生产环境的应用程序

模式：基于能力（6 个独立的服务模块）

核心原则（经 AWS 于 2025 年 12 月验证）：

1. 与框架无关——适用于任意 Agent 框架或模型
2. 零基础设施——完全托管，无运维开销
3. 会话隔离——会话间完全隔离数据
4. 企业级安全——VPC、PrivateLink、身份集成
5. 可组合服务——按需使用
6. 可用于生产环境——为规模化、可靠性和安全性而构建

质量目标：

• 部署：从代码到生产环境用时 < 5 分钟
• 延迟：支持低延迟工作负载和长达 8 小时的异步工作负载
• 可观测性：全面集成 CloudWatch

---

适用场景

以下情况可使用 bedrock-agentcore：

• 在 AWS 上构建生产环境 AI Agent
• 需要托管基础设施来部署 Agent
• 需要会话隔离和企业级安全
• 希望使用现有 Agent 框架（LangGraph、CrewAI 等）
• 需要浏览器自动化或代码执行功能
• 与现有身份提供方集成

不适用场景：

• 简单的 Bedrock 模型调用（使用 bedrock-runtime）
• 使用操作组的标准 Bedrock Agents（使用 bedrock-agent）
• 非 AWS 部署

---

前提条件

必需

• 拥有可访问 Bedrock 的 AWS 账户
• AgentCore 服务的 IAM 权限
• Python 3.10+（用于 SDK）

建议

• 已安装 bedrock-agentcore-sdk-python
• bedrock-agentcore-starter-toolkit CLI
• 已启用基础模型访问权限（Claude 等）

安装

# Install SDK and CLI
pip install bedrock-agentcore strands-agents bedrock-agentcore-starter-toolkit

# Verify installation
agentcore --help

---

核心服务

1. AgentCore Runtime

用于运行 Agent 代码的安全、会话隔离计算环境。

Boto3 客户端：

import boto3

# Data plane operations
client = boto3.client('bedrock-agentcore')

# Control plane operations
control = boto3.client('bedrock-agentcore-control')

创建 Agent Runtime：

# Using starter toolkit
# agentcore configure -e main.py -n my-agent
# agentcore deploy

# Using boto3 control plane
response = control.create_agent_runtime(
    name='my-production-agent',
    description='Customer service agent',
    agentRuntimeArtifact={
        's3': {
            'uri': 's3://my-bucket/agent-package.zip'
        }
    },
    roleArn='arn:aws:iam::123456789012:role/AgentCoreExecutionRole',
    pythonRuntime='PYTHON_3_13',
    entryPoint=['main.py']
)
agent_runtime_arn = response['agentRuntimeArn']

调用 Agent：

# Invoke deployed agent
response = client.invoke_agent_runtime(
    agentRuntimeArn='arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/xxx',
    runtimeSessionId='session-123',
    payload={
        'prompt': 'What is my order status?',
        'context': {'user_id': 'user-456'}
    }
)

result = response['payload']
print(result)

Agent 入口点结构：

from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent

app = BedrockAgentCoreApp(debug=True)
agent = Agent()

@app.entrypoint
def invoke(payload):
    """Main agent entry point"""
    user_message = payload.get("prompt", "Hello!")
    app.logger.info(f"Processing: {user_message}")

    result = agent(user_message)
    return {"result": result.message}

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

---

2. AgentCore Gateway

将现有 API 和 Lambda 函数转换为兼容 Agent 的工具，并支持通过语义搜索发现工具。

创建 Gateway：

response = control.create_gateway(
    name='customer-service-gateway',
    description='Gateway for customer service tools',
    protocolType='REST'
)
gateway_arn = response['gatewayArn']

添加 Gateway 目标（工具）：

# Add an existing Lambda as a tool
response = control.create_gateway_target(
    gatewayId='gateway-xxx',
    name='GetOrderStatus',
    description='Retrieves order status by order ID',
    targetConfiguration={
        'lambdaTarget': {
            'lambdaArn': 'arn:aws:lambda:us-east-1:123456789012:function:GetOrder'
        }
    },
    toolSchema={
        'name': 'get_order_status',
        'description': 'Get the current status of a customer order',
        'inputSchema': {
            'type': 'object',
            'properties': {
                'order_id': {
                    'type': 'string',
                    'description': 'The unique order identifier'
                }
            },
            'required': ['order_id']
        }
    }
)

同步工具：

# Sync gateway tools for discovery
control.synchronize_gateway_targets(
    gatewayId='gateway-xxx'
)

---

3. 浏览器 Runtime

安全地执行复杂的 Web 工作流。

启动浏览器会话：

response = client.start_browser_session(
    browserId='browser-xxx',
    sessionConfiguration={
        'timeout': 300,
        'viewport': {'width': 1920, 'height': 1080}
    }
)
session_id = response['browserSessionId']

执行浏览器操作：

# Navigate and interact
response = client.update_browser_stream(
    browserSessionId=session_id,
    action={
        'navigate': {'url': 'https://example.com'},
        'click': {'selector': '#submit-button'},
        'type': {'selector': '#search', 'text': 'query'}
    }
)

---

4. 代码解释器

安全地执行代码，以完成数据分析和可视化等任务。

启动代码解释器会话：

response = client.start_code_interpreter_session(
    codeInterpreterId='interpreter-xxx'
)
session_id = response['codeInterpreterSessionId']

执行代码：

response = client.invoke_code_interpreter(
    codeInterpreterSessionId=session_id,
    code='''
import pandas as pd
import matplotlib.pyplot as plt

# Analyze data
df = pd.DataFrame({'x': [1,2,3,4,5], 'y': [2,4,6,8,10]})
plt.plot(df['x'], df['y'])
plt.savefig('output.png')
print(df.describe())
''',
    language='PYTHON'
)

output = response['output']
files = response['files']  # Generated files

---

5. 身份集成

与现有身份提供方原生集成，以实现身份验证和权限委派。

创建 OAuth2 提供方：

response = control.create_oauth2_credential_provider(
    name='okta-provider',
    credentialProviderVendor='OKTA',
    oauth2ProviderConfig={
        'clientId': 'your-client-id',
        'clientSecret': 'your-client-secret',
        'authorizationServerUrl': 'https://your-domain.okta.com/oauth2/default',
        'scopes': ['openid', 'profile', 'email']
    }
)

创建工作负载身份：

response = control.create_workload_identity(
    name='agent-identity',
    allowedRoleArns=['arn:aws:iam::123456789012:role/AgentRole']
)

获取访问令牌：

# Get token for workload
response = client.get_workload_access_token(
    workloadIdentityId='identity-xxx'
)
access_token = response['accessToken']

---

6. 可观测性

通过 CloudWatch 和 OpenTelemetry 实现实时可见性。

启用可观测性：

# In your agent entry point
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider

# Configure tracing
provider = TracerProvider()
trace.set_tracer_provider(provider)

# Entry point with OTel
# entryPoint: ['opentelemetry-instrument', 'main.py']

CloudWatch 指标：

• 每个会话的令牌用量
• 延迟（p50、p95、p99）
• 会话时长
• 错误率
• 工具调用成功率/失败率

---

Boto3 客户端参考

数据平面（bedrock-agentcore）

方法	用途
invoke_agent_runtime	执行 Agent 逻辑
stop_runtime_session	停止活动会话
list_sessions	列出所有会话
start_browser_session	初始化浏览器
stop_browser_session	结束浏览器会话
start_code_interpreter_session	启动解释器
invoke_code_interpreter	执行代码
batch_create_memory_records	创建记忆
retrieve_memory_records	获取记忆
get_workload_access_token	获取身份验证令牌
evaluate	运行评估

Control Plane（bedrock-agentcore-control）

方法	用途
create_agent_runtime	创建运行时
delete_agent_runtime	移除运行时
update_agent_runtime	修改运行时
create_gateway	创建网关
create_gateway_target	添加工具
create_memory	创建记忆存储
create_policy	创建策略
create_evaluator	创建评估器
create_browser	创建浏览器
create_code_interpreter	创建解释器

---

快速入门：Hello World Agent

第 1 步：创建 Agent 文件

# main.py
from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent

app = BedrockAgentCoreApp()
agent = Agent(model="anthropic.claude-sonnet-4-20250514-v1:0")

@app.entrypoint
def invoke(payload):
    prompt = payload.get("prompt", "Hello!")
    result = agent(prompt)
    return {"response": result.message}

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

第 2 步：配置并部署

# Configure
agentcore configure -e main.py -n hello-world-agent

# Test locally
python main.py &
curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Hello!"}'

# Deploy to AWS
agentcore deploy

# Test deployed
agentcore invoke '{"prompt": "Hello from production!"}'

第 3 步：以编程方式调用

import boto3

client = boto3.client('bedrock-agentcore')

response = client.invoke_agent_runtime(
    agentRuntimeArn='arn:aws:bedrock-agentcore:us-east-1:123456789012:agent-runtime/hello-world',
    runtimeSessionId='test-session-1',
    payload={'prompt': 'What can you help me with?'}
)

print(response['payload'])

---

错误处理

from botocore.exceptions import ClientError

try:
    response = client.invoke_agent_runtime(
        agentRuntimeArn=agent_arn,
        runtimeSessionId='session-1',
        payload={'prompt': 'test'}
    )
except ClientError as e:
    error_code = e.response['Error']['Code']

    if error_code == 'ResourceNotFoundException':
        print("Agent runtime not found")
    elif error_code == 'ValidationException':
        print("Invalid request parameters")
    elif error_code == 'ThrottlingException':
        print("Rate limited - implement backoff")
    elif error_code == 'AccessDeniedException':
        print("Check IAM permissions")
    else:
        raise

---

IAM 权限

最小执行角色

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "bedrock-agentcore:InvokeAgentRuntime",
        "bedrock-agentcore:StartBrowserSession",
        "bedrock-agentcore:InvokeCodeInterpreter"
      ],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "bedrock:InvokeModel",
        "bedrock:InvokeModelWithResponseStream"
      ],
      "Resource": "arn:aws:bedrock:*::foundation-model/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    }
  ]
}

---

相关技能

• bedrock-agentcore-policy：Cedar 策略编写与强制执行
• bedrock-agentcore-evaluations：Agent 测试与质量评估
• bedrock-agentcore-memory：情景记忆与短期记忆管理
• bedrock-agentcore-deployment：生产部署模式
• bedrock-agentcore-multi-agent：多 Agent 编排（A2A 协议）
• boto3-eks：适用于托管在 EKS 上的 Agent
• terraform-aws：基础设施即代码

---

参考资料

• references/gateway-configuration.md - 网关配置详情
• references/identity-integration.md - OAuth 与工作负载身份
• references/troubleshooting.md - 常见问题与解决方案

---

来源

• Amazon Bedrock AgentCore
• Boto3 BedrockAgentCore
• AgentCore SDK Python
• AgentCore Starter Toolkit