用 AWS Agent Registry 在运行时发现工具

AWS ML Blog 2026-10-21T21:41:31.062081

用 AWS Agent Registry 做运行时发现，可以解决一个 agent 演示阶段容易藏起来的问题。工具少的时候，硬编码完全没问题；到几十个就开始吃力；再多就管不住了。企业系统会暴露几百个内部 API，涉及定价、客服、订单、库存和客户数据。在这种规模下，模型花在辨别工具名字上的 token，可能比理解用户请求还多。每加一个新能力，orchestrator 就得改代码、重新部署，还要跨团队协调发布，而各团队其实更想按自己的节奏发版。团队把大量内部 API 开放给 agent 使用时，这种模式在企业软件即服务（SaaS）项目里很常见。工具目录一扩大，构建期就把所有工具接好，反而成了瓶颈。

一个思路是改变 orchestrator 获取工具的方式：不再把每个工具都编译进 agent，而是让 agent 在运行时自己找相关工具。AWS Agent Registry 是一个托管服务，用来编目和发现模型上下文协议（MCP）服务器、agent 和 agent 技能。发布者注册协议相关的描述符和可搜索的描述。运行时，orchestrator 可以搜索已批准、可发现的记录，用返回的端点和描述符内容来构建客户端。Registry 的搜索结果不会返回能力的调用凭证或认证方式。

这篇文章会走一遍重构后的 Python 示例演示的「发现 → 实例化 → 执行」循环。代码通过 Amazon Bedrock AgentCore Gateway（Amazon Bedrock AgentCore 的一项能力）部署一个订单管理 MCP 服务器，还在 AgentCore 运行时上部署两个 A2A（agent 间通信）agent，再加一个 Registry 和一个 orchestrator。示例目录里有部署脚本、orchestrator、辅助模块和清理脚本。

解决方案概览：发现、实例化、执行

这个示例里的 orchestrator agent 不内置任何工具或 agent 的知识。

图 1：没有 AWS Agent Registry 时（左），端点都是硬编码的，新增一个 agent 就得改代码再重新部署。用了 AWS Agent Registry 之后（右），编排器通过语义搜索发现能力，新集成的组件只要注册并通过审批就能直接用。

发现

编排器向 AWS Agent Registry 的数据面发送一条自然语言查询。搜索结果里包含记录标识、名称、描述、记录类型、状态、版本，以及协议描述符，比如 MCP 服务器元数据或 A2A agent card。结果中不含调用凭证。应用必须通过另一条可信的配置通道，自行获取并应用相应的凭证。

实例化

对每一条返回的记录，编排器根据描述符类型构建对应的客户端。在本示例中，注册的 MCP 服务器代表的是先被添加为 Amazon Bedrock AgentCore Gateway 目标的那些工具，因此 MCP 描述符指向的是 Gateway 端点。编排器另外获取 Cognito OAuth 访问令牌，在创建 MCPClient 时使用。

A2A 记录包含一个 agent card，其中的 URL 指向某个 AgentCore 运行时端点。封装层通过 boto3 数据面客户端调用该运行时，请求会用调用方的 AWS 凭证签名。注册中心发现提供的是连接元数据，而不是认证材料。

执行

系统会构建一个临时的内部 agent，只带上动态发现的工具。大语言模型（LLM）为请求挑选合适的工具，结果再汇总返回给用户。这一轮结束后，内部 agent 及其客户端随即销毁。下一次请求会重新搜索，拿到一套全新的工具。

图 2：端到端架构。Amazon Bedrock AgentCore 运行时上的一个编排器查询 AWS Agent Registry。匹配到的 MCP 服务器通过 Amazon Bedrock AgentCore Gateway 调用，匹配到的 A2A agent 则在 Amazon Bedrock AgentCore 运行时上调用。

示例采用的参考架构包括 AWS Agent Registry、Amazon Bedrock AgentCore Gateway、Amazon Bedrock AgentCore 运行时、Amazon Bedrock、Amazon Cognito、AWS Lambda 和 AWS Secrets Manager。

三个演示场景用不同的能力组合来验证发现流程。第一个是纯 MCP 调用（查询订单状态）。第二个把 MCP 和 A2A 串联起来（定价与折扣）。第三个针对模糊的用户意图做纯语义匹配：退货与退款，而所匹配的 A2A 智能体名称里并不包含这两个词。

技术实现

配套的运行时发现示例会部署依赖、创建并填充 Registry、验证发现能力、部署编排器，然后运行三个演示。orchestrator_agent.py 包含每次请求触发的发现循环。utils.py 负责解析 Registry 描述符，并创建 MCP 和 A2A 客户端。cleanup.py 用于清理资源。

1. 注册模型

一条 Registry 记录包含可搜索的描述，以及一份协议相关的描述符。示例用 Agent Registry 控制平面客户端完成 Registry 和记录的生命周期操作，用 Agent Registry 数据平面客户端完成运行时发现。

下面创建一个需要手动审批、使用 IAM 授权的 Registry：

registry_client = session.client("agent-registry-control")
dp_client = session.client("agent-registry")
reg = registry_client.create_registry(
    name="OrderManagementRegistry",
    description=(
        "Registry for Order Management & Customer Service — "
        "agent discovers tools and agents via semantic search"
    ),
    approvalConfiguration={"autoApprovalRules": []},
)
REGISTRY_ARN = reg["registryArn"]
REGISTRY_ID = REGISTRY_ARN.split("/")[-1]

创建记录时必须提供 recordType 字段。在这个示例中，MCP 服务器使用 recordType="MCP"，基于 A2A 的 agent 使用 recordType="AGENT"。协议相关的元数据存放在 descriptors 对象里的 mcpServer 或 a2aAgentCard 下。

MCP 描述符包含服务器元数据和工具定义。A2A 描述符包含 agent 卡片。描述符里放的是连接信息，但不包含调用方在调用该能力时必须提供的凭证。

下面按照当前公开 SDK 的结构，分别构造 MCP 和 A2A 记录的描述符。

# MCP 描述符
mcp_descriptors = {
    "mcpServer": {
        "data": json.dumps(server_metadata),
        "dataSchemaVersion": "2025-12-11",
        "additionalData": {
            "tools": {
                "data": json.dumps({"tools": tool_schemas}),
                "dataSchemaVersion": "2025-06-18",
            }
        },
    }

# A2A 描述符
a2a_descriptors = {
    "a2aAgentCard": {
        "data": json.dumps(agent_card),
        "dataSchemaVersion": "0.3",
    }

描述要写成用户能看懂的语言，并让这条记录和相邻记录区分开来。描述符内容提供协议相关的服务器、工具或 agent 卡片数据。搜索质量应该用有代表性的查询来评估。本文并不认为目录规模变大，排名就一定会变好。

下面是按当前控制面结构创建 Registry 记录的示例。

descriptors = (
    mcp_descriptors
    if cfg["protocol"] == "MCP"
    else a2a_descriptors
)

response = registry_client.create_registry_record(
    registryId=REGISTRY_ID,
    name=name,
    description=cfg["description"],
    recordType=cfg["protocol"],
    descriptors=descriptors,
    recordVersion="1.0",
)
record_id = response["recordId"]

2. 审批工作流

把 autoApproval 设为 false 后，新记录必须先从 DRAFT 变为 PENDING_APPROVAL，再变为 APPROVED。提交和更新状态这两个操作都受 IAM 授权控制，因此可以把发布者权限和审批者权限分开管理。Registry 的可发现数据平面只会返回已批准的记录。

两步式审批状态流转。

record = registry_client.get_registry_record(
    registryId=REGISTRY_ID,
    recordId=record_id,
)
if record["status"] == "DRAFT":
    registry_client.submit_registry_record_for_approval(
        registryId=REGISTRY_ID,
        recordId=record_id,
    )
    registry_client.update_registry_record_status(
        registryId=REGISTRY_ID,
        recordId=record_id,
        status="APPROVED",
        statusReason="Approved",
    )

3. 发现层

数据平面客户端支持对可发现记录做关键词与语义混合搜索。最新的方法名是 search_discoverable_registry_records。返回结果会暴露规范化后的记录字段和可发现描述符，但不包含能力本身的调用凭证。

在多个协议下搜索已批准的可发现记录。

response = dp_client.search_discoverable_registry_records(
    registryIds=[REGISTRY_ARN],
    searchQuery=query,
    maxResults=3,
)
records = response.get("registryRecords", [])

在示例场景中，用「退货退款」相关的查询能召回 Customer Support Agent，因为它的描述里体现了这一意图。这只是演示结果，不能据此断言：只要目录规模变大，排序效果就一定会变好。

4. 编排器

编排器是一个部署在 Amazon Bedrock AgentCore runtime 上的 Strands agent。它对外提供 discover_and_execute 方法：先搜索 Registry，解析每个返回的描述符，创建对应的 MCP 或 A2A 客户端，然后只用这次发现到的能力，拼出一个仅在当前请求内有效的内部 agent。

图 3：编排器的单次请求流程。外部 agent 把用户请求转换成一次 Registry 搜索。内部 agent 由搜索结果组装而成，并在本轮结束时销毁。

discover_and_execute 的核心逻辑可以浓缩在一个代码块里：

编排器的发现/实例化/执行循环。

@tool
def discover_and_execute(request: str) -> str:
    records = dp_client.search_discoverable_registry_records(
        registryIds=[REGISTRY_ARN],
        searchQuery=request,
        maxResults=5,
    ).get("registryRecords", [])

# 注册表返回的是描述符，不是调用凭证。
    oauth_token = fetch_oauth_token_from_config()
    mcp_clients, a2a_tools = [], []
    for record in records:
        metadata = parse_server_metadata(record)
        if metadata["protocol"] == "MCP":
            mcp_clients.append(
                create_mcp_client_from_metadata(metadata, oauth_token)
            )
        elif metadata["protocol"] == "A2A":
            a2a_tools.append(
                create_a2a_tool_from_metadata(metadata, session, REGION)
            )

return run_request_scoped_agent(request, mcp_clients, a2a_tools)

    model = BedrockModel(model_id=MODEL_ID, region_name=REGION)
    started_clients = []
    try:
        mcp_tools = []
        for c in mcp_clients:
            c.start()
            started_clients.append(c)
            mcp_tools.extend(c.list_tools_sync())

## agent compiled in realtime within agent
        sub_agent = Agent(
            model=model,
            tools=mcp_tools + a2a_tools,
            system_prompt=(
                "You are an Order Management & Customer Service assistant. "
                "Use the RIGHT tool for each request:\n"
                "- Order lookups, status, cancellations, address changes: use MCP tools\n"
                "- Pricing, discounts, promo codes: use the pricing_agent tool\n"
                "- Returns, refunds, complaints, escalations: use the customer_support_agent tool\n"
                "Always use the most specific tool for the task."
            ),
        )
        return str(sub_agent(request))
    finally:
        for c in started_clients:
            try:
                c.stop()
            except Exception:
                pass

这个示例会并行发出多条搜索查询，以提高订单管理这一领域的召回率，对记录去重，并在请求结束后停掉 MCP 客户端。这种并行是示例自己的编排选择，并不是 Registry 的硬性要求。认证不在发现环节里：Gateway 的 OAuth 令牌来自 Cognito 配置和 Secrets Manager，而 A2A 运行时调用则通过 boto3 会话签名。

5. 跑三个示例

每个示例对应发现循环的一种形态。

示例 1：订单状态（只用 MCP）

这个请求最终只落到一次 MCP 工具调用。

result = invoke_orchestrator(
    "What is the current status and tracking info for order 123?",
    agentcore_client, orchestrator_arn,
)

Registry 返回订单管理的 MCP 服务器。编排器通过 Gateway 调用 get_order_status，LLM 把结果组织成一段人能读懂的回答，内部 agent 随即结束。这一轮完全不涉及 A2A。

示例 2：定价与折扣（MCP + A2A）

这个请求最终落到一条跨协议的链路。

result = invoke_orchestrator(
    "Order 123 has 2x Widget Pro at $99.98. What discount tiers or "
    "promo codes can reduce the price?",
    agentcore_client, orchestrator_arn,
)

Registry 同时返回订单管理的 MCP 服务器和定价的 A2A agent：两种协议，一句自然语言查询。LLM 先调用 get_order_status 拿到订单明细，再在 runtime 上调用定价 agent 分析折扣，最后汇总成一条建议。

示例 3：退货与退款（语义匹配）

这个请求的意图，在任何工具名称里都找不到字面对应。

result = invoke_orchestrator(
    "I want to return order 789 (Premium Headphones, delivered "
    "March 10). Am I eligible for a return and what is the refund "
    "amount?",
    agentcore_client, orchestrator_arn,
)

这个示例体现了语义发现的价值。「return refund」这个说法并没有出现在 Customer Support Agent 注册的名称里。它之所以排名靠前，是因为它的描述抓住了这个意图。没有 Registry 的话，编排器就必须硬编码「哪个 agent 负责退货」这样的知识。

6. 可观测性

用 Amazon Bedrock AgentCore Observability 和 Amazon CloudWatch（在合适的地方）为发现和调用这条链路加上监控。值得关注的信号包括：Registry 搜索的成功与失败、选中的记录标识符、客户端构建错误、下游调用结果、端到端链路关联，以及应用错误。示例授予了 Amazon CloudWatch Logs 权限，但没有创建自定义指标或仪表盘。生产团队应根据自己的运维模式定义所需的追踪、日志和告警。生产部署还应把 Amazon Bedrock Guardrails 作为 agent 回复的标准控制手段。

受治理的运行时发现给团队带来什么

发现循环搭好之后，原先跟编排器发版绑在一起的几项能力，就都独立出来了。

什么时候不该用运行时发现

不是所有工作负载都需要运行时发现。以下两种场景仍然更适合在构建期显式接线：

结论

运行时从 Registry 发现资产，让编排器不再那么依赖一份固定目录。面对 AWS Agent Registry，编排器可以用语义加关键词的混合搜索查找已审批的能力，解析 MCP 或 A2A 描述符，并创建请求级的客户端。治理决定哪些东西可以被发现。认证则始终是 Registry 和每个被调用能力各自需要明确处理的事。

AWS Agent Registry 已经正式可用，其能力超出这个单账号示例：

评测应针对组织自己的目录，验证搜索相关性和调用行为。

这个示例体现了语义发现的价值。「return refund」这个说法并没有出现在 Customer Support Agent 注册的名称里。它之所以排名靠前，是因为它的描述抓住了这个意图。没有 Registry 的话，编排器就必须硬编码「哪个 agent 负责退货」这样的知识。

6. 可观测性

用 Amazon Bedrock AgentCore Observability 和 Amazon CloudWatch（在合适的地方）为发现和调用这条链路加上监控。值得关注的信号包括：Registry 搜索的成功与失败、选中的记录标识符、客户端构建错误、下游调用结果、端到端链路关联，以及应用错误。

示例授予了 Amazon CloudWatch Logs 权限，但没有创建自定义指标或仪表盘。生产团队应根据自己的运维模式定义所需的追踪、日志和告警。生产部署还应把 Amazon Bedrock Guardrails 作为 agent 回复的标准控制手段。

受治理的运行时发现给团队带来什么

发现循环搭好之后，原先跟编排器发版绑在一起的几项能力，就都独立出来了。

什么时候不该用运行时发现

不是所有工作负载都需要运行时发现。以下两种场景仍然更适合在构建期显式接线：

查看原文