包教会！本地构建一个实用AI Agent

Amazon Bedrock AgentCore简介

Amazon Bedrock AgentCore是一套专为AI Agent设计的企业级基础设施服务，解决Agent从原型到生产过程中的核心挑战：

• AgentCore Runtime：Serverless运行时环境，支持最长8小时任务执行和100MB请求负载；
• AgentCore Memory：提供短期与长期记忆管理，支持跨会话上下文保持；
• AgentCore Gateway：统一工具网关，支持MCP协议的工具发现和调用；
• AgentCore Identity：安全的身份认证与授权管理。

本地开发具备显著优势：开发者可在熟悉环境中快速验证想法，无需等待云端部署，同时保持与生产环境一致的代码结构。

第一部分：环境准备

1.1 安装Python包管理器

建议使用uv作为Python包管理器，具备更快的依赖解析速度和更可靠的环境隔离能力。

macOS和Linux系统：

curl -LsSf https://astral.sh/uv/install.sh | sh

Windows系统：

powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

安装完成后，请重新打开终端以使环境变量生效。

1.2 配置亚马逊云科技凭证

AgentCore需访问Amazon Bedrock服务，因此需配置有效的亚马逊云科技凭证。

# 验证当前凭证配置
aws sts get-caller-identity

若尚未配置凭证，请执行：

aws configure

系统将提示输入Access Key ID、Secret Access Key、默认区域和输出格式。

1.3 开启模型访问权限

在亚马逊云科技控制台中，进入Amazon Bedrock服务，在“Model access”页面启用Anthropic Claude系列模型访问权限。本文示例使用Claude Sonnet 4模型。

1.4 创建项目并安装依赖

# 创建新项目，指定 Python 版本为 3.13
uv init my-agent-project --python 3.13

# 进入项目目录
cd my-agent-project

# 安装核心依赖包
uv add bedrock-agentcore strands-agents

# 安装开发工具包（用于后续部署）
uv add --dev bedrock-agentcore-starter-toolkit

依赖包说明

• bedrock-agentcore：AgentCore的Python SDK，提供运行时封装和服务集成；
• strands-agents：Strands Agent框架，简化Agent构建过程；
• bedrock-agentcore-starter-toolkit：部署工具包，提供CLI命令和自动化部署能力。

第二部分：构建第一个本地Agent

2.1 创建Agent入口文件

在项目根目录下创建main.py文件：

from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent
from strands.models import BedrockModel

app = BedrockAgentCoreApp(debug=True)

model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
    temperature=0.7,
    max_tokens=4096
)

agent = Agent(
    model=model,
    system_prompt="你是一个专业的技术助手，回答问题时准确、简洁。"
)

@app.entrypoint
def invoke(payload):
    user_message = payload.get("prompt", "")
    if not user_message:
        return {"error": "prompt 参数不能为空"}
    result = agent(user_message)
    return {"response": result.message["content"][0]["text"]}

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

2.2 代码逐行解析

导入模块

from bedrock_agentcore import BedrockAgentCoreApp

BedrockAgentCoreApp是AgentCore应用的核心类，封装HTTP服务器、请求路由和生命周期管理，使Agent代码可在本地或云端标准化运行。

from strands import Agent

Agent是Strands框架核心类，负责管理对话流程、工具调用和模型交互，提供简洁API构建具备推理能力的AI Agent。

from strands.models import BedrockModel

BedrockModel是Bedrock模型封装类，处理与Amazon Bedrock服务的认证、请求格式化和响应解析。

创建应用实例

app = BedrockAgentCoreApp(debug=True)

创建AgentCore应用实例。debug=True启用调试模式，输出详细请求与响应日志，便于开发阶段排查问题；生产环境建议设为False。

配置模型

model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
    temperature=0.7,
    max_tokens=4096
)

• model_id：指定基础模型，此处为Claude Sonnet 4，“us.”前缀表示使用美国区域跨区域推理端点；
• temperature：控制输出随机性（0–1），较低值更确定，较高值更富创造性；
• max_tokens：限制单次响应最大token数，防止输出过长。

创建Agent

agent = Agent(
    model=model,
    system_prompt="你是一个专业的技术助手，回答问题时准确、简洁。"
)

• model：传入已配置的模型实例；
• system_prompt：系统提示词，定义Agent角色与行为准则，每次对话起始时发送给模型，引导其行为。

定义入口函数

@app.entrypoint
def invoke(payload):

@app.entrypoint装饰器将invoke函数注册为Agent入口点。收到HTTP请求后，AgentCore自动调用此函数，并将请求体解析为payload字典传入。

user_message = payload.get("prompt", "")

从请求负载提取prompt字段，使用get方法并设默认值，可安全处理字段缺失。

if not user_message:
    return {"error": "prompt 参数不能为空"}

输入验证：确保用户提供了有效输入，返回字典将被自动序列化为JSON响应。

result = agent(user_message)

调用Agent处理用户消息，Agent会将消息发送至模型、获取响应，并在需要时执行工具调用；返回result对象含完整对话结果。

return {"response": result.message["content"][0]["text"]}

从结果中提取文本响应并返回。result.message是模型返回的消息对象，content是内容数组，取首元素text字段。

启动应用

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

Python标准入口点检查。当直接运行该文件时，启动HTTP服务器。app.run()默认在0.0.0.0:8080启动服务，为AgentCore Runtime标准端口。

2.3 运行和测试

启动Agent

打开终端，在项目目录下执行：

uv run main.py

您将看到类似以下输出：

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8080

发送测试请求

打开另一终端窗口，使用curl发送请求：

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "请解释什么是微服务架构"}' | jq .

运行结果示例

{
  "response": "微服务架构是一种软件设计模式，它将大型应用程序拆分为多个小型、独立的服务，每个服务负责特定的业务功能。\n\n### 核心特征\n\n**1. 服务独立性**\n- 每个微服务可以独立开发、部署和扩展\n- 拥有自己的数据库和业务逻辑\n- 通过 API 进行通信\n\n**2. 单一职责**\n- 每个服务专注于一个业务领域\n- 服务边界清晰，功能内聚\n\n**3. 去中心化**\n- 没有中央控制器\n- 服务间通过轻量级协议通信（如 HTTP/REST、消息队列）\n\n### 主要优势\n- **可扩展性**：可针对性地扩展高负载服务\n- **技术多样性**：不同服务可使用不同技术栈\n- **故障隔离**：单个服务故障不会影响整个系统\n- **开发效率**：小团队可独立维护服务\n- **部署灵活**：支持持续集成和部署\n\n### 主要挑战\n- **复杂性增加**：网络通信、数据一致性问题\n- **运维成本**：需要更多的监控和管理工具\n- **性能开销**：服务间调用的网络延迟\n- **分布式系统问题**：如分布式事务、服务发现\n\n微服务架构特别适合大型、复杂的企业应用，以及需要快速迭代和高可用性的系统。"
}

响应显示Agent成功接收请求并返回结构化回答，响应时间约9秒（含模型推理时间）。

请求解析

• -X POST：指定HTTP方法为POST；
• http://localhost:8080/invocations：AgentCore标准调用端点；
• -H "Content-Type: application/json"：设置请求头表明请求体为JSON格式；
• -d '{"prompt": "..."}'：请求体，含发送给Agent的消息；
• | jq .：通过jq工具格式化输出（需预先安装）。

第三部分：为Agent添加工具能力

Agent的真正威力在于调用外部工具完成任务。以下将为Agent添加自定义工具。

3.1 安装工具依赖

uv add strands-agents-tools

strands-agents-tools包含预构建工具（如计算器、文件操作等）。

3.2 创建带工具的Agent

更新main.py文件：

from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent, tool
from strands.models import BedrockModel
from strands_tools import calculator

app = BedrockAgentCoreApp(debug=True)

@tool
def get_weather(city: str) -> str:
    """
    获取指定城市的当前天气信息。

    Args:
        city: 城市名称，例如 "北京"、"上海"、"深圳"

    Returns:
        包含温度和天气状况的字符串
    """
    weather_data = {
        "北京": "晴天，气温 22°C，湿度 45%",
        "上海": "多云，气温 26°C，湿度 65%",
        "深圳": "阵雨，气温 28°C，湿度 80%",
    }
    return weather_data.get(city, f"暂无 {city}的天气数据")

@tool
def search_database(query: str, limit: int = 5) -> str:
    """
    在数据库中搜索相关信息。

    Args:
        query: 搜索关键词
        limit: 返回结果的最大数量，默认为 5

    Returns:
        搜索结果的摘要信息
    """
    return f"找到 {limit}条与「{query}」相关的记录（模拟数据）"

model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514-v1:0"
)

agent = Agent(
    model=model,
    tools=[calculator, get_weather, search_database],
    system_prompt="""你是一个智能助手，具备以下能力：
1. 使用 calculator 工具进行数学计算
2. 使用 get_weather 工具查询城市天气
3. 使用 search_database 工具搜索数据库

请根据用户的问题，选择合适的工具来提供准确的答案。"""
)

@app.entrypoint
def invoke(payload):
    user_message = payload.get("prompt", "")
    if not user_message:
        return {"error": "prompt 参数不能为空"}
    result = agent(user_message)
    return {"response": result.message["content"][0]["text"]}

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

3.3 工具定义详解

@tool 装饰器

@tool
def get_weather(city: str) -> str:

@tool装饰器将普通Python函数转换为Agent可调用工具。Strands框架自动提取函数签名作为输入参数定义、解析docstring作为描述信息、处理参数类型转换与验证。

类型注解的重要性

def get_weather(city: str) -> str:

• city: str：参数类型注解，指示该参数应为字符串；
• -> str：返回值类型注解，表明函数返回字符串。

类型注解不仅是代码文档，更是模型理解工具用法的关键依据，模型据此构造工具调用参数。

Docstring规范

"""
获取指定城市的当前天气信息。

Args:
    city: 城市名称，例如 "北京"、"上海"、"深圳"

Returns:
    包含温度和天气状况的字符串
"""

Docstring含三部分：
• 首行：工具简要描述，模型依此判断是否调用该工具；
• Args：各参数详细说明，含示例值；
• Returns：返回值描述。
清晰Docstring能显著提升模型选择和使用工具的准确性。

带默认值的参数

def search_database(query: str, limit: int = 5) -> str:

limit: int = 5定义带默认值的可选参数。模型可选择是否提供此参数，未提供则使用默认值5。

注册工具到Agent

agent = Agent(
    model=model,
    tools=[calculator, get_weather, search_database],
    ...
)

tools参数接受工具列表。calculator为strands_tools内置工具，get_weather与search_database为自定义工具。

3.4 测试工具调用

重启Agent后，测试不同请求：

测试计算功能

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "计算 (125 + 375) * 2.5 的结果"}' | jq .

Agent识别为数学计算问题，自动调用calculator工具。

计算功能运行结果

{
  "response": "计算结果：(125 + 375) * 2.5 = 1250\n\n计算过程：\n1. 先计算括号内：125 + 375 = 500\n2. 再乘以 2.5：500 * 2.5 = 1250"
}

测试天气查询

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "上海今天天气怎么样？"}' | jq .

Agent调用get_weather工具，传入参数city="上海"。

天气查询运行结果

{
  "response": "上海今天的天气情况：多云，气温 26°C，湿度 65%。\n\n温度适中，湿度稍高，建议外出时携带一把伞以防万一。"
}

测试复合任务

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"prompt": "查一下北京的天气，如果温度超过20度，计算开10小时空调需要多少度电（假设功率1.5千瓦）"}' | jq .

该请求需Agent协调多个工具：先查天气获取温度，再依条件决定是否计算。

复合任务运行结果

{
  "response": "让我先查询北京的天气，然后根据温度情况进行计算。\n\n**北京天气**：晴天，气温 22°C，湿度 45%\n\n由于当前温度 22°C 超过了 20 度，我来计算空调用电量：\n\n**计算过程**：\n- 空调功率：1.5 千瓦（kW）\n- 使用时间：10 小时\n- 耗电量 = 功率 × 时间 = 1.5 kW × 10 h = 15 度电\n\n**结论**：开 10 小时空调大约需要消耗 15 度电。按照一般民用电价（约 0.5 元/度）计算，费用约为 7.5 元。"
}

结果表明Agent成功：1）调用get_weather获取北京天气；2）判断温度（22°C）超20度；3）调用calculator计算用电量；4）综合信息给出完整回答。

第四部分：构建MCP Server

Model Context Protocol（MCP）是标准化协议，允许Agent动态发现和调用工具。MCP Server可被任何支持MCP的客户端（如Claude Desktop、Cursor、Kiro等）调用。

4.1 安装MCP依赖

uv add mcp

4.2 创建MCP Server

创建新文件mcp_server.py：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    name="Financial Tools Server",
    host="0.0.0.0",
    stateless_http=True
)

@mcp.tool()
def add_numbers(a: int, b: int) -> int:
    """Add two numbers together"""
    return a + b

@mcp.tool()
def multiply_numbers(a: int, b: int) -> int:
    """Multiply two numbers together"""
    return a * b

@mcp.tool()
def get_stock_price(symbol: str) -> str:
    """
    Get the current stock price for a given symbol.

    Args:
        symbol: Stock ticker symbol like AAPL, GOOGL, AMZN

    Returns:
        Stock symbol and current price
    """
    prices = {
        "AAPL": 178.50,
        "GOOGL": 141.20,
        "AMZN": 178.90,
        "MSFT": 378.50,
    }
    price = prices.get(symbol.upper())
    if price:
        return f"{symbol.upper()} current price: ${price}"
    return f"Price data not found for {symbol}"

@mcp.tool()
def greet_user(name: str) -> str:
    """Greet a user by name"""
    return f"Hello, {name}! Nice to meet you."

if __name__ == "__main__":
    print("MCP Server 启动中...")
    print("MCP 端点: http://localhost:8000/mcp")
    mcp.run(transport="streamable-http")

4.3 MCP Server代码解析

导入FastMCP

from mcp.server.fastmcp import FastMCP

FastMCP是MCP Python SDK提供的高级封装类，简化MCP Server创建。

创建MCP Server实例

mcp = FastMCP(
    name="Financial Tools Server",
    host="0.0.0.0",
    stateless_http=True
)

• name：Server名称，用于标识和日志；
• host：监听地址，“0.0.0.0”表示接受所有网络接口连接；
• stateless_http=True：启用无状态HTTP模式，为AgentCore Runtime要求配置。

定义工具

@mcp.tool()
def add_numbers(a: int, b: int) -> int:
    """Add two numbers together"""
    return a + b

@mcp.tool()装饰器将Python函数注册为MCP工具：
• 函数名成为工具名称；
• 参数类型注解定义输入类型；
• Docstring成为工具描述；
• 返回值类型注解定义输出类型。

启动服务

mcp.run(transport="streamable-http")

transport="streamable-http"指定Streamable HTTP传输协议，为MCP推荐生产环境方式，支持流式响应。

4.4 运行和测试MCP Server

启动服务

uv run mcp_server.py

运行输出

MCP Server 启动中...
MCP 端点: http://localhost:8000/mcp
INFO:     Started server process [12345]
INFO:     Uvicorn running on http://0.0.0.0:8000

使用Python客户端测试

首先安装MCP客户端依赖（如未安装）：

uv add mcp httpx

创建测试客户端文件mcp_client.py：

import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def main():
    """测试 MCP Server 的连接和工具调用"""
    mcp_url = "http://localhost:8000/mcp"

    print(f"连接到 MCP Server: {mcp_url}")

    async with streamablehttp_client(mcp_url) as (read_stream, write_stream, _):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            print("✓ 连接成功\n")

            # 列出可用工具
            tools = await session.list_tools()
            print("可用工具:")
            for tool in tools.tools:
                print(f" - {tool.name}: {tool.description}")

            print("\n" + "=" * 50 + "\n")

            # 测试 add_numbers 工具
            result = await session.call_tool("add_numbers", {"a": 10, "b": 20})
            print(f"add_numbers(10, 20) = {result.content[0].text}")

            # 测试 multiply_numbers 工具
            result = await session.call_tool("multiply_numbers", {"a": 7, "b": 8})
            print(f"multiply_numbers(7, 8) = {result.content[0].text}")

            # 测试 get_stock_price 工具
            result = await session.call_tool("get_stock_price", {"symbol": "AAPL"})
            print(f"get_stock_price('AAPL') = {result.content[0].text}")

            # 测试 greet_user 工具
            result = await session.call_tool("greet_user", {"name": "开发者"})
            print(f"greet_user('开发者') = {result.content[0].text}")

if __name__ == "__main__":
    asyncio.run(main())

代码解析

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

导入MCP客户端所需模块：
• ClientSession：MCP客户端会话管理类；
• streamablehttp_client：Streamable HTTP传输客户端，用于连接该协议MCP Server。

async with streamablehttp_client(mcp_url) as (read_stream, write_stream, _):

建立与MCP Server连接，返回读写流用于双向通信。

async with ClientSession(read_stream, write_stream) as session:
    await session.initialize()

创建客户端会话并初始化，与服务器交换能力信息。

tools = await session.list_tools()

获取服务器提供的所有工具列表。

result = await session.call_tool("add_numbers", {"a": 10, "b": 20})

调用指定工具，传入参数字典，返回执行结果。

运行测试

确保MCP Server运行，执行：

uv run mcp_client.py

运行结果

连接到 MCP Server: http://localhost:8000/mcp
✓ 连接成功

可用工具:
 - add_numbers: Add two numbers together
 - multiply_numbers: Multiply two numbers together
 - get_stock_price: Get the current stock price for a given symbol.
 - greet_user: Greet a user by name

==================================================

add_numbers(10, 20) = 30
multiply_numbers(7, 8) = 56
get_stock_price('AAPL') = AAPL current price: $178.5
greet_user('开发者') = Hello, 开发者! Nice to meet you.

使用curl测试（CLI方式）

MCP Server使用Streamable HTTP传输协议，需设置正确Accept header：

# 初始化连接
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
      "protocolVersion": "2024-11-05",
      "capabilities": {},
      "clientInfo": {"name": "curl-client", "version": "1.0.0"}
    }
  }'

# 列出可用工具
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}'

# 调用 add_numbers 工具
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 3,
    "method": "tools/call",
    "params": {"name": "add_numbers", "arguments": {"a": 10, "b": 20}}
  }'

curl响应示例

列出工具响应：

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "tools": [
      {"name": "add_numbers", "description": "Add two numbers together", ...},
      {"name": "multiply_numbers", "description": "Multiply two numbers together", ...},
      {"name": "get_stock_price", "description": "Get the current stock price for a given symbol.", ...},
      {"name": "greet_user", "description": "Greet a user by name", ...}
    ]
  }
}

调用工具响应：

{"jsonrpc": "2.0", "id": 3, "result": {"content": [{"type": "text", "text": "30"}]}}

注意：curl命令必须含-H "Accept: application/json, text/event-stream" header，否则返回“Not Acceptable”错误。

使用MCP Inspector测试（可选）

MCP Inspector是可视化测试工具，提供图形界面测试MCP Server：

npx @modelcontextprotocol/inspector

浏览器打开http://localhost:6274，选“Streamable HTTP”，输入MCP Server地址http://localhost:8000/mcp，即可交互测试：
• 查看所有工具及其参数定义；
• 手动调用工具并查看结果；
• 调试工具执行过程。

第五部分：构建A2A Server

Agent-to-Agent（A2A）协议专为多Agents系统设计，支持Agent间相互发现与通信。A2A使用JSON-RPC格式通信，并通过Agent Card描述Agent能力。

5.1 安装A2A依赖

uv add 'strands-agents[a2a]'

5.2 创建A2A Server

创建新文件a2a_server.py：

import os
import logging
from strands import Agent
from strands.multiagent.a2a import A2AServer
from strands.models import BedrockModel
from strands_tools import calculator
import uvicorn
from fastapi import FastAPI

logging.basicConfig(level=logging.INFO)

# 从环境变量获取运行时 URL，本地开发时使用默认值
runtime_url = os.environ.get(
    'AGENTCORE_RUNTIME_URL',
    'http://127.0.0.1:9000/'
)
logging.info(f"Runtime URL: {runtime_url}")

# 创建模型
model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514-v1:0"
)

# 创建 Agent
strands_agent = Agent(
    name="Calculator Agent",
    description="A calculator agent that can perform basic arithmetic operations.",
    model=model,
    tools=[calculator],
    callback_handler=None
)

# 创建 A2A Server
a2a_server = A2AServer(
    agent=strands_agent,
    http_url=runtime_url,
    serve_at_root=True  # 在根路径提供服务
)

# 创建 FastAPI 应用
app = FastAPI(title="Calculator A2A Server")

@app.get("/ping")
def health_check():
    """健康检查端点"""
    return {"status": "healthy"}

# 挂载 A2A Server
app.mount("/", a2a_server.to_fastapi_app())

if __name__ == "__main__":
    print("A2A Server 启动中...")
    print("Agent Card: http://localhost:9000/.well-known/agent-card.json")
    print("健康检查: http://localhost:9000/ping")
    uvicorn.run(app, host="0.0.0.0", port=9000)

5.3 A2A Server代码解析

导入A2A模块

from strands.multiagent.a2a import A2AServer

A2AServer是Strands SDK提供的A2A协议实现，需安装strands-agents[a2a]扩展包。

环境变量配置

runtime_url = os.environ.get(
    'AGENTCORE_RUNTIME_URL',
    'http://127.0.0.1:9000/'
)

从环境变量读取运行时URL，未设置则用本地默认值。该设计使同一代码可在本地与云端运行，仅通过