在 AWS 上用 A2A 构建可组合、可互操作的多智能体系统

跨越组织边界的多智能体系统中的智能体，通常需要以可组合、可互操作的方式与由其他组织构建和运行的智能体协作。它们需要一种方式来相互发现、通信和交换任务，同时不暴露内部实现，也不要求双方使用相同的框架。

智能体到智能体（A2A）协议解决了这一约束。没有 A2A，跨组织边界构建可组合的多智能体系统需要为每个客户和框架编写自定义集成代码。独立软件供应商（ISV）则可以构建可组合、可互操作的智能体，与客户已经在使用的兼容 A2A 的框架协同工作。

虽然本文以 ISV 用例为框架展开，但这一模式同样适用于跨边界暴露智能体的组织，无论是面向合作伙伴、客户，还是使用不同框架的内部团队。随着系统的增长，你可以通过向编排器注册 Agent Card 来添加技能，而无需更改已经在运行的智能体。

在本文中，我们将介绍一个部署在 AWS 上的参考架构。基于 LangChain 的企业编排器使用 A2A 跨组织边界与基于 Strands Agents 的软件即服务（SaaS）智能体通信，企业和 SaaS 提供商位于不同的 AWS 账户中。双方不依赖对方的框架或 SDK，也不共享任何机密。我们将介绍架构和实现模式，然后从三个 Jupyter notebook 部署该系统并运行端到端测试。

A2A 协议现已成为 Linux 基金会的项目，AWS 是支持该协议的组织之一。借助 A2A，智能体可以通过标准化协议相互发现、协商任务并跨组织边界协作。该协议使用 OAuth 2.0 和 OpenID Connect（OIDC）等现有身份验证标准来实现跨边界安全，并被设计为可跨兼容 A2A 的框架工作。

在 Open Protocols for Agent Interoperability Part 4: Inter-Agent Communication on A2A 一文中，我们介绍了 A2A 的概念，并演示了两个 Strands 智能体在 AWS 上使用该协议进行通信。那篇文章通过 Agent Card 发现、结构化任务执行和 A2A 服务器模式奠定了基础。本文在此基础上进一步探讨跨框架互操作性、跨组织信任边界以及双向 OAuth。

解决方案概述

三个智能体运行在两个 AWS 账户中，一个属于企业，一个属于 SaaS 提供商。A2A 协议将它们连接起来，并使用双向 OAuth 身份验证。每个账户都有自己的 Amazon Cognito 用户池，因此每一方都使用自己的身份提供程序进行身份验证。

图 1：跨企业账户和 SaaS 提供商账户的 A2A 智能体

在入驻过程中，企业和 SaaS 提供商只交换公开的值，例如身份提供程序 URL、客户端 ID、作用域和端点 URL。随后，双方都会信任来自对方用户池的令牌，且不共享任何机密。

入驻完成后，每个请求遵循以下步骤：

企业应用从企业的 Amazon Cognito 用户池获取访问令牌。
该应用通过 HTTPS 将问题发送到 Amazon API Gateway 中的用户 API，然后轮询答案。
Amazon API Gateway 通过 VPC 链接和内部网络负载均衡器，将请求私密地转发给 Enterprise Orchestrator——一个 LangChain 智能体。
当其模型判定该问题需要进行合规扫描时，编排器会从企业用户池中获取自己的令牌，所用客户端密钥存储在 AWS Secrets Manager 中。
编排器向 Amazon Bedrock AgentCore 运行时上的合规 SaaS 智能体发送 A2A 请求，使用的是它在启动时从该智能体的 Agent Card 中读取到的技能。
合规智能体在注册时登记的回调 URL 处检查企业上下文智能体的 Agent Card，然后从 SaaS 提供商自己的用户池获取令牌。
合规智能体回调企业上下文智能体以获取账户元数据，完成扫描，并将答案返回给编排器，再返回给应用。

在企业侧，两个智能体运行在 Amazon Elastic Container Service（Amazon ECS）和 AWS Fargate 上。Enterprise Orchestrator 是用户的入口。LangChain 提供智能体框架和工具调用抽象，Amazon Bedrock 则提供驱动每个智能体推理的基础模型（FM）。编排器使用 LangChain 和 Amazon Bedrock（Anthropic Claude Sonnet 4.6）。启动时，编排器从 SaaS 智能体的知名 URL 获取其 Agent Card，并为卡片中描述的每个技能创建一个 LangChain 工具。随后模型根据用户的问题决定调用哪个工具，因此编排器无需硬编码的路由逻辑。它通过 A2A 并使用 OAuth 持有者令牌发送合规扫描请求，并为管理层汇总结果。用户通过一个仅接受企业自身用户池令牌的 Amazon API Gateway HTTPS 端点访问编排器。

第二个企业智能体提供内部上下文。Enterprise Context Agent 使用 LangChain 和 Amazon Bedrock（Amazon Nova Lite）构建。它提供内部账户元数据，例如内部策略规则、数据分类、业务重要性以及升级联系人（团队值班轮换）。所部署的智能体使用官方 a2a-sdk（AgentCard、AgentExecutor 以及符合规范的 JSON-RPC 路由），并固定在 a2a-sdk[http-server]==1.2.2 版本。SaaS 智能体通过一个仅接受 SaaS 提供商回调令牌的 API Gateway 合作伙伴 API 访问上下文智能体。

两个企业智能体都运行在 Amazon Virtual Private Cloud（Amazon VPC）的私有子网中，没有公共 IP 地址。

在 SaaS 提供商侧，Compliance SaaS Agent 运行在 SaaS 提供商自己的 AWS 账户中的 Amazon Bedrock AgentCore 上。它使用 Strands Agents SDK 构建，根据企业自身的内部策略检查 AWS 账户，例如生产环境中不得有公开存储，以及机密数据必须静态加密。在生产环境中，扫描工具会连接到 AWS Config、AWS Security Hub 或 Amazon GuardDuty 等服务。在本演示中，该智能体使用模拟跨多个账户合规发现的合成数据。

在开始扫描之前，SaaS 智能体会回调 Enterprise Context Agent 以获取企业上下文，例如哪些账户属于生产环境、适用哪些内部政策、以及应向谁上报。这种双向模式使两个智能体成为真正的对等方，而非客户端与服务器的关系。回调只会发送到客户在入驻时注册的地址。

关键的设计决策是所有跨边界通信都使用 A2A。企业无法访问 SaaS 智能体的内部工具、系统提示词或模型选择。这种分离意味着 ISV 可以在不更改客户集成的情况下更改其内部架构，只要其 Agent Card 和 A2A 接口保持兼容即可。SaaS 卡片只发布一项经过精选的技能，而不是其内部工具名称。

Amazon Bedrock 通过 API 提供对多个基础模型的访问，因此每个智能体都可以使用适合其工作负载的模型。在本解决方案中，编排器和 SaaS 智能体使用 Anthropic Claude Sonnet 4.6 进行复杂推理，而上下文智能体使用 Amazon Nova Lite 进行元数据查询。您可以根据自身需求选择不同的模型。

A2A 与 Model Context Protocol (MCP) 在此架构中解决的是不同的问题。MCP 以客户端-服务器模型将智能体连接到其自身的工具和数据源。A2A 则以对等方式将智能体连接到其他智能体，双方都不暴露其内部实现。在生产环境中，SaaS 智能体可以在内部使用 MCP 连接到其合规性扫描工具，同时通过 A2A 向外部暴露自身。无论智能体如何连接其内部工具，外部接口都保持不变。

先决条件

这些 Notebook 包含完整的部署流程。在运行它们之前，您需要满足以下条件：

两个 AWS 账户，一个用于企业，一个用于 SaaS 提供商，且位于相同的 AWS 区域：us-east-1、us-east-2 或 us-west-2。在两个账户中，都需要能够访问 Amazon Bedrock 中的 Anthropic Claude Sonnet 4.6；在企业账户中，还需要能够访问 Amazon Nova Lite。在首次运行之前，请在每个账户中提交 Anthropic 首次使用表单，并为运行 Notebook 所用的身份授予 Amazon Bedrock 模型访问文档中列出的 AWS Marketplace 权限。模型可用性因区域而异。请查看 Amazon Bedrock 模型可用性文档以获取当前列表。
A Jupyter environment with Python 3.10 or later, either:
- 在您的计算机上：JupyterLab、AWS Command Line Interface (AWS CLI) 版本 2，以及为每个账户设置的、通过 aws configure sso 配置的 AWS IAM Identity Center 配置文件。您不需要 Docker：容器镜像由 AWS CodeBuild 构建。
- 或者，在每个账户中使用一个 Amazon SageMaker AI Notebook 实例（ml.t3.medium 即可）。请参阅 Amazon SageMaker AI 入门设置指南。
在每个账户中，创建一个具有创建 AWS CloudFormation 堆栈及其中资源权限的 AWS Identity and Access Management (IAM) 角色。每个 Notebook 在其 Reference 部分列出了所需的确切权限。请参阅 IAM 用户指南。

Notebook 3 还会运行本文稍后描述的端到端测试，以便您验证部署。

运行此解决方案会为它所创建的资源产生费用。在企业账户中，主要的常驻资源包括一个 NAT 网关、两个 AWS Fargate 任务、内部 Network Load Balancer 以及两个接口 VPC 终端节点。Amazon Bedrock 和 Amazon Bedrock AgentCore 按使用量计费。价格因 AWS 区域而异。请参阅 Amazon VPC、AWS Fargate、Amazon Bedrock、Amazon Bedrock AgentCore 和 Elastic Load Balancing 的定价页面。完成操作后请执行 Clean up 部分。

实施方案

使用三个 notebook 部署该系统，按顺序通过 Jupyter 逐个单元格运行。Notebook 1（企业账户，约 3 分钟）创建企业身份并写入接入请求。Notebook 2（SaaS 提供商账户，约 5 分钟）部署合规智能体并写入接入响应。Notebook 3（企业账户，约 20 分钟）部署企业智能体并运行测试。接入文件中仅包含公开值。

克隆 GitHub 仓库，并为本地运行设置两个 AWS 配置文件，或者为每个账户设置一个 notebook 实例，并将仓库克隆到每个实例中。
在企业账户中运行 notebook 1。
将 onboarding-request.json 提供给 SaaS 提供商。
在 SaaS 提供商账户中运行 notebook 2。
将 onboarding-response.json 返还给企业。
在企业账户中运行 notebook 3。
测试完成后，先运行 notebook 3 的 Clean up，再运行 notebook 2 的。

以下部分将解释该部署背后的关键实现模式。它们共同展示了三项能力：

跨框架互操作性 – LangChain 编排器使用 A2A 调用 Strands 智能体，双方互不依赖对方的 SDK。
跨组织边界的可组合性 – ISV 的合规智能体与企业的上下文智能体组合成一个双方都无法独立构建的工作流。
基于 OAuth 的双向协作 – 智能体以对等身份行动，通过基于令牌的身份验证在信任边界之间互相调用。

借助 A2A，该场景中的 ISV 用单一的基于协议的接口替代了多条自定义集成路径。

构建 SaaS 合规智能体

使用 Strands Agents SDK 构建 ISV 的合规智能体，并将其部署到 Amazon Bedrock AgentCore。在本演示中，扫描工具返回的是合成的合规结果。

该智能体通过 Strands 的 @tool 装饰器暴露三个合规工具（列出可用账户、运行合规扫描和检索修复步骤），外加一个回调企业的第四个工具，将在下一部分介绍。Agent Card 并不列出这些工具。它只发布一个精心策划的技能，这样调用方的模型看到的是产品做什么，而不是它是如何构建的。这个简化示例展示了扫描工具：

from strands import Agent, ToolContext, tool
from strands.models.bedrock import BedrockModel
from strands.multiagent.a2a import A2AServer

@tool
def run_compliance_scan(account_id: str) -> dict:
    """Run a compliance scan on a specific AWS account.

    Args:
        account_id: The AWS account ID to scan (e.g., acc-12345)
    """
    if account_id not in ACCOUNTS:
        return {"error": f"Account {account_id} not found."}
    findings = COMPLIANCE_FINDINGS.get(account_id, [])
    non_compliant = [f for f in findings if f["status"] == "NON_COMPLIANT"]
    return {
        "account_id": account_id,
        "account_name": ACCOUNTS[account_id]["name"],
        "scan_summary": {
            "total_rules_checked": len(findings),
            "compliant": len(findings) - len(non_compliant),
            "non_compliant": len(non_compliant),
            "critical_findings": len([f for f in non_compliant if f["severity"] == "CRITICAL"]),
            "high_findings": len([f for f in non_compliant if f["severity"] == "HIGH"]),
        },
        "findings": findings,
    }

启用双向回调

这正是该解决方案超越典型智能体集成之处。大多数智能体对智能体的演示只展示单向的请求和响应：客户端智能体调用服务器端智能体并等待结果。当被调用的智能体需要只有调用方才拥有的信息时，这种模式就行不通了。在生产环境中，智能体需要成为可以互相调用的对等方。

SaaS 合规智能体需要企业上下文。例如哪些账户属于生产环境、适用哪些内部策略，以及应向谁上报升级。这些是只有企业才拥有的信息。

该 SaaS 智能体通过 A2A 回调，但只回调客户在入驻时注册的回调 URL。它首先在那里验证客户的公开上下文卡片，然后从 SaaS 提供商自己的 Cognito 用户池获取令牌，并向企业合作方 API 发送 message/send：

def _call_enterprise_context(message_text: str, inbound_task_id: str = "-") -> str:
    if not CALLBACK_URL_VALID: # checked once at import
        raise CallbackUnavailable("CALLBACK_URL is not a registered callback URL")
    cfg = _load_callback_config() # this SaaS provider's own client, from Secrets Manager
    _verify_context_card(cfg) # anonymous GET of the card at CALLBACK_URL, no redirects
    rpc_id = f"req-{uuid4().hex}"
    payload = {
        "jsonrpc": "2.0",
        "id": rpc_id,
        "method": "message/send",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"kind": "text", "text": message_text}],
                "messageId": uuid4().hex,
            }
        },
    }

    for attempt in (1, 2):
        token = _fetch_token(cfg, force=attempt > 1) # from the SaaS provider's own Cognito pool
        try:
            resp = _post_callback(CALLBACK_URL, payload, token)
            ...

每一次失败——从未经验证的卡片到被拒绝的令牌——都只会给模型一条消息：“企业上下文不可用”，而详细信息则保留在 SaaS 提供商的日志中。

在 AgentCore 上通过 A2A 公开智能体

Strands SDK 中的 A2AServer 类负责处理 A2A 协议、Agent Card 生成以及 JSON-RPC 消息处理。该服务器发布一个精选技能，并在卡片上声明调用者如何获取令牌：

a2a_server = SecuredA2AServer( # A2AServer plus the card's securitySchemes
    agent_factory=create_compliance_agent,
    http_url=AGENT_URL,
    serve_at_root=True,
    skills=[COMPLIANCE_SCAN_SKILL], # one curated skill, not the internal tools
    enable_a2a_compliant_streaming=True,
)

部署到 AgentCore 后，它会自动在 /.well-known/agent-card.json 处生成一个 Agent Card。Agent Card 是一个 JSON 文档，描述智能体的名称、能力和技能。兼容 A2A 的客户端可以获取此卡片，无需文档或 SDK 集成即可了解该智能体能做什么。

用 LangChain 构建企业编排器

这是互操作性的证明。该企业编排器使用 LangChain，特意与基于 Strands 的 SaaS 智能体不同。它使用原生 httpx HTTP 调用和 A2A JSON-RPC，不依赖 Strands SDK。如果 ISV 从 Strands 切换到另一个兼容 A2A 的框架，只要 SaaS 智能体仍然讲 A2A 并发布兼容的 Agent Card，企业编排器无需任何更改即可继续工作。这就是协议级互操作性在实践中意味着什么。

编排器从企业自己的 Cognito 用户池获取令牌，并且仅通过 HTTPS 向固定为获取 Agent Card 所用地址的智能体 URL 发送 JSON-RPC 载荷。这段节选代码展示了核心 A2A 客户端：

from langchain_aws import ChatBedrock
from langchain_core.tools import tool as langchain_tool
from langchain.agents import create_agent

llm = ChatBedrock(
    model_id="us.anthropic.claude-sonnet-4-6",
    region_name=AWS_REGION,
    max_tokens=4096,
    config=BEDROCK_CLIENT_CONFIG, # timeouts and standard retries
    guardrails=GUARDRAIL_CONFIG, # this account's guardrail on every model call
)

def send_a2a_message(agent_url: str, message_text: str) -> str:
    # agent_url is the pinned agent URL from discovery: the address the Agent Card
    # was fetched from, never a value from a caller or a message.
    rpc_id = f"lc-{uuid4().hex}"
    payload = {
        "jsonrpc": "2.0",
        "id": rpc_id,
        "method": "message/send",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"kind": "text", "text": message_text}],
                "messageId": uuid4().hex,
            }
        },
    }
    def _post(token: str, attempt: int) -> httpx.Response:
        headers = {"Content-Type": "application/json", "Authorization": f"Bearer {token}"}
        with httpx.Client(timeout=_A2A_TIMEOUT) as client: # Timeout(300.0, connect=5.0); httpx doesn't follow redirects by default
            resp = client.post(agent_url, json=payload, headers=headers)
            resp.raise_for_status()
            return resp

    # get_bearer_token() raises when there is no valid client config and never returns
    # an empty token, so no request goes out without one (fail closed).
    token = get_bearer_token()
    try:
        resp = _post(token, 1)
    except httpx.HTTPStatusError as e:
        # On a 401, force one token refresh and retry once.
        if e.response.status_code == 401:
            resp = _post(_fetch_token(force=True), 2)
        else:
            raise
    ...

用 LLM 发现智能体并路由请求

启动时，编排器从配置的 ISV URL 获取 Agent Card。它从每张卡片读取技能描述，并为每个技能创建一个 LangChain 工具。编排器的大语言模型（LLM）Anthropic Claude Sonnet 4.6 随后根据用户的问题决定调用哪个工具。这段简化代码展示了该模式：

def _tool_name_part(text: str) -> str:
    """Reduce card text to the characters Bedrock allows in a tool name (letters, digits, _)."""
    return re.sub(r"[^A-Za-z0-9_]+", "_", str(text)).strip("_") or "agent"

def build_langchain_tools(agents: list[dict]) -> list:
    """Create a LangChain tool for each skill in each discovered agent."""
    tools = []

    for agent in agents:
        agent_url = agent["url"]
        agent_name = _tool_name_part(agent["name"])

        for skill in agent["skills"]:
            skill_name = _tool_name_part(skill["name"])
            skill_description = skill.get("description", "")
            examples = skill.get("examples", [])

            # Build a descriptive tool name and description
            tool_name = f"{agent_name}__{skill_name}"[:64]
            tool_description = (
                f"{skill_description} "
                f"(External agent: {agent['name']})"
            )
            if examples:
                tool_description += f" Examples: {'; '.join(examples[:3])}"

            # Create a closure to capture agent_url for this specific tool
            def make_tool(name: str, desc: str, url: str):
                def call_external_agent(question: str) -> str:
                    """Send a natural language question to an external agent via A2A protocol."""
                    return send_a2a_message(url, question)
                call_external_agent.__name__ = name
                call_external_agent.__doc__ = desc
                return langchain_tool(call_external_agent)

            tools.append(make_tool(tool_name, tool_description, agent_url))

    # (A catch-all tool for an agent whose card lists no skills is omitted here.)
    return tools

agent_tools = build_langchain_tools(discovered_agents) # from the Agent Cards fetched at startup
agent_executor = create_agent(llm, agent_tools, system_prompt=SYSTEM_PROMPT)

这样，当 ISV 添加或更改某项能力时，其 Agent Card 会随之变化，编排器在下次重启时会拾取新技能。两个团队无需协调这一变更。

构建企业上下文智能体

企业上下文智能体使用 LangChain 和 Amazon Nova Lite 进行企业元数据查询，并通过官方 a2a-sdk 经由 A2A 公开其能力。该 SDK 负责 Agent Card 服务、JSON-RPC 路由和协议一致性。由于 A2A 是基于 HTTP 的 JSON-RPC，标准 Web 框架即可提供服务。编排器展示了客户端：它使用纯 httpx 发送 A2A 请求，不使用 A2A SDK。

该智能体还在 /.well-known/agent-card.json 处发布自己的 Agent Card，SaaS 智能体在回调之前会检查它。这与另一个方向使用的发现机制相同。A2A 规范定义了 Agent Card，因此它们不依赖于构建智能体的框架。

使用 Amazon Cognito 设置跨边界 OAuth

跨越组织边界的智能体间通信如果没有身份验证，大多数企业是无法接受的。该解决方案中的每个 A2A 任务请求（message/send）都携带 JSON Web Token（JWT）。双方各自使用来自自己身份提供商的令牌进行身份验证，而接收方只接受该提供商、一个客户端和一个作用域。唯一未经身份验证的端点是企业上下文智能体的 Agent Card，它发布描述性元数据，供调用方发现该智能体。

该解决方案使用 Amazon Cognito 作为 OAuth 提供商。在更新示例中特定于 Cognito 的检查之后，同样的模式也可以使用其他签发已签名 JWT 访问令牌的 OpenID Connect（OIDC）身份提供商。

身份验证流程涉及四个组件：

Amazon Cognito：每个账户一个用户池。企业池向编排器和企业应用颁发令牌，SaaS 池向 SaaS 智能体颁发令牌用于其回调。你可以在 Amazon Cognito 控制台中查看它们。

Amazon Bedrock AgentCore 配置了企业池的发现 URL、编排器的客户端 ID 和一个作用域。它在每个传入的 A2A 请求上验证 JWT：

{"customJWTAuthorizer": {
    "discoveryUrl": "https://cognito-idp.us-east-1.amazonaws.com/<enterprise pool ID>/.well-known/openid-configuration",
    "allowedClients": ["<a2a-enterprise-orchestrator-saas client ID>"],
    "allowedScopes": ["a2a-enterprise-compliance-saas/invoke"]
}}

Amazon API Gateway 使用 JWT 授权方。合作伙伴 API 只接受 SaaS 提供商的回调令牌，用户 API 只接受企业应用的令牌。可在 Amazon API Gateway 控制台中查看配置。

每个账户中的 AWS Secrets Manager 存储该账户的客户端密钥，并通过 IAM 授权和拒绝所有其他主体的资源策略，将每个密钥限制为仅由使用它的那一个工作负载访问。没有任何密钥被硬编码或与另一个账户共享。你可以在 AWS Secrets Manager 控制台中管理密钥。

以下 AWS CloudFormation 片段展示了合作伙伴 API 的 JWT 授权方：

# Trusts only the SaaS provider's pool, its callback client, and its callback scope.
ContextApiAuthorizer:
  Type: AWS::ApiGatewayV2::Authorizer
  Properties:
    ApiId: !ImportValue
      Fn::Sub: '${IdentityStackName}-PartnerApiId'
    AuthorizerType: JWT
    Name: saas-jwt
    IdentitySource:
      - '$request.header.Authorization'
    JwtConfiguration:
      Audience: [!Ref SaasCallbackClientId]
      Issuer: !Ref SaasIssuer
ContextApiRoute:
  Type: AWS::ApiGatewayV2::Route
  Properties:
    RouteKey: 'POST /'
    AuthorizationType: JWT
    AuthorizerId: !Ref ContextApiAuthorizer
    AuthorizationScopes: [!Ref SaasCallbackScope]
    # ApiId and Target omitted

在 API Gateway 与 AgentCore 之间配置 OAuth 时，机器对机器身份验证的一个细节是常见的错误来源。Amazon Cognito 的客户端凭证访问令牌携带 client_id 和 scope，但没有 aud 声明。当 aud 缺失时，API Gateway 的 JWT 授权方会对照受众列表检查 client_id。在 AgentCore 一侧，配置 allowedClients 并保持 allowedAudience 为未设置状态。

使用笔记本部署系统

这些笔记本通过 Jupyter 完成整个部署。它们部署五个 AWS CloudFormation 堆栈，三个在企业账户中，两个在 SaaS 提供商账户中。企业基础设施堆栈与企业服务堆栈之所以分开，是因为存在先有鸡还是先有蛋的问题：ECS 服务需要 Docker 镜像，但保存这些镜像的 Amazon Elastic Container Registry（Amazon ECR）存储库是由 CloudFormation 创建的。你可以在 AWS CloudFormation 控制台中监控这些堆栈。

这些笔记本会尽早暴露故障。每个笔记本在部署任何内容之前，都会先用一次简短的测试调用检查 Amazon Bedrock 模型访问权限，并且每一步都会在出现第一个错误时停止并给出清晰的输出。Clean up（清理）部分位于笔记本 2 和笔记本 3 的底部，位于一个停止单元格之后，因此 Run All 永远不会执行到它们。

了解 Agent Card 与发现机制

Agent Card（智能体卡片）让发现成为可能，而无需事先就工具定义达成一致。没有它们，企业需要文档、SDK，或者与 ISV 的工程团队开会，才能了解合规智能体能做什么。有了 Agent Card，企业编排器只需获取一份 JSON 文档，即可在运行时以程序化方式立即知道智能体的名称、能力和技能。这是该合规智能体为某客户发布的卡片（有删节）：

{
    "name": "Compliance SaaS Agent",
    "description": "Scans AWS accounts for compliance with your organization's policies and returns findings by severity with remediation steps.",
    "url": "https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/.../invocations/",
    "protocolVersion": "0.3.0",
    "capabilities": { "streaming": true },
    "skills": [
        {
            "id": "compliance-scan",
            "name": "Compliance scan",
            "description": "Scans your AWS accounts against your organization's policies and returns findings by severity, remediation steps with estimated effort, and your escalation contacts when available."
        }
    ],
    "securitySchemes": {
        "cognito": {
            "type": "oauth2",
            "description": "OAuth 2.0 client credentials from the customer's identity provider, registered at onboarding.",
            "flows": { "clientCredentials": {
                "tokenUrl": "https://a2a-enterprise-<account-id>.auth.us-east-1.amazoncognito.com/oauth2/token",
                "scopes": { "a2a-enterprise-compliance-saas/invoke": "Invoke the compliance agent" }
            } }
        }
    },
    "security": [ { "cognito": ["a2a-enterprise-compliance-saas/invoke"] } ]
}

每个技能的描述都会成为工具的 docstring，因此编排器的模型可将每个请求路由到正确的智能体。两个团队无需就工具定义达成一致，也无需共享 SDK。卡片还声明了调用方如何进行身份验证。编排器会核对该声明而非盲目遵循：只有当卡片中写明的令牌端点和作用域与入驻时注册的一致时，才会保留该卡片。

端到端测试

Notebook 3 通过两个账户运行一次端到端测试。它为企业应用获取令牌，提交一个问题，然后轮询结果：

# A token for the enterprise's own user client. The secret is read from this account's
# Cognito pool in memory and discarded; it is never printed or typed.
secret = cognito.describe_user_pool_client(
    UserPoolId=ENT_POOL_ID, ClientId=ORCH_USER_CLIENT_ID)["UserPoolClient"]["ClientSecret"]
r = requests.post(TOKEN_ENDPOINT, auth=(ORCH_USER_CLIENT_ID, secret), allow_redirects=False, timeout=15,
                  data={"grant_type": "client_credentials", "scope": "a2a-enterprise-orchestrator/ask"})
del secret
auth = {"Authorization": f"Bearer {r.json()['access_token']}"}

# Submit (HTTP 202 and a job ID), then poll every 5 seconds
r = requests.post(f"{ORCH_API_URL}/jobs", headers=auth, allow_redirects=False, timeout=30,
                  json={"question": "Scan our production accounts for compliance with our internal "
                                    "policies, and include escalation contacts "
                                    "for critical findings."})
job_id = r.json()["job_id"]
while (job := requests.get(f"{ORCH_API_URL}/jobs/{job_id}", headers=auth,
                           allow_redirects=False, timeout=30).json())["status"] not in ("succeeded", "failed"):
    time.sleep(5)
del auth
print(job["response"])

编排器通过其 Agent Card 发现 SaaS 智能体，并通过 A2A 路由请求。SaaS 智能体回调企业上下文智能体获取账户元数据（双向 A2A），使用其内部工具运行合规扫描，将所有内容合并为一份增强报告，然后发送回去。响应中包含来自两个智能体的数据。这表明两个独立构建的智能体，运行在不同框架和不同基础设施上，可以组合成两者都无法单独产出的工作流。

只有当回复中包含合规发现、包含只有企业上下文智能体知道的值班轮换代码，且合作伙伴 API 的访问日志显示 SaaS 回调客户端收到 200 时，测试才算通过。这些共同证明回调成功了。Notebook 3 随后运行七项信任边界检查，向每个关卡发送错误、缺失或伪造的令牌，并且每次调用都必须被拒绝。以下是一次测试运行的输出摘录（ID 为占位符）：

[22:53:27] INFO Step E11 End-to-end test: submit a job and poll it (usually about 1 to 3 min; waits up to 10) ...
[22:53:27] INFO POST /jobs: HTTP 202 (apigw-requestid=<request-id>)
[22:53:27] INFO job_id=<job-id>
[22:53:27] INFO job status: running
[22:54:08] INFO job status: succeeded
------------------------------------------------------------------
[22:54:19] INFO partner API access log: 28/Oct/2026:22:53:31 +0000 POST / status=200 clientId=<callback-client-id> requestId=<request-id>
[22:54:19] INFO [ok ] Compliance findings (SaaS agent)
[22:54:19] INFO [ok ] Context-only marker (context agent, through the callback)
[22:54:19] INFO [ok ] Partner API access log: 200 for POST / from the SaaS callback client
[22:54:19] INFO PASS E11: job <job-id> has compliance findings and a context-only marker, and the partner API logged the callback
[22:54:19] INFO done in 52.5s

[22:54:51] INFO Step E12 Trust boundary checks: each token works only where it's trusted ...
[22:54:52] INFO 1 orchestrator-saas token on the partner API POST /: 401 as expected (www-authenticate='Bearer scope="a2a-saas-customer-callback/invoke" error="invalid_token" error_description="unable to find an appropriate key in the JWKS. ..."', apigw-requestid=<request-id>, x-amzn-requestid=-)
[22:54:52] INFO 2 orchestrator-user token on the partner API POST /: 401 as expected (www-authenticate='Bearer scope="a2a-saas-customer-callback/invoke" error="invalid_token" error_description="unable to find an appropriate key in the JWKS. ..."', apigw-requestid=<request-id>, x-amzn-requestid=-)
[22:54:52] INFO 3a orchestrator-user token on the SaaS runtime Agent Card: 401 as expected (www-authenticate='Bearer resource_metadata="...", scope="a2a-enterprise-compliance-saas/invoke"', apigw-requestid=-, x-amzn-requestid=<request-id>)
[22:54:53] INFO 3b orchestrator-user token on the SaaS runtime message/send: 401 as expected (www-authenticate='Bearer resource_metadata="...", scope="a2a-enterprise-compliance-saas/invoke"', apigw-requestid=-, x-amzn-requestid=<request-id>)
[22:54:53] INFO 4 no token on POST /jobs: 401 as expected (www-authenticate='Bearer', apigw-requestid=<request-id>, x-amzn-requestid=-)
[22:54:53] INFO 5 orchestrator-saas token on the user API POST /jobs: 401 as expected (www-authenticate='Bearer scope="a2a-enterprise-orchestrator/ask" error="invalid_token" error_description="the token does not have a valid audience"', apigw-requestid=<request-id>, x-amzn-requestid=-)
[22:54:53] INFO 6 forged SaaS-issuer token (invalid signature) on the partner API POST /: 401 as expected (www-authenticate='Bearer scope="a2a-saas-customer-callback/invoke" error="invalid_token" error_description="unable to find an appropriate key in the JWKS. ..."', apigw-requestid=<request-id>, x-amzn-requestid=-)
[22:54:53] INFO PASS E12: all seven trust boundary checks passed; each token works only where it's trusted
[22:54:53] INFO done in 1.4s

在该次运行中，报告覆盖了两个生产账户和六项发现，并包含每个账户的升级联系人。你可以从任何 HTTP 客户端发送同样的请求，无需 A2A SDK。

生产环境注意事项

为求简洁，本演示对若干方面做了简化。在生产环境中，请考虑以下改动：

每个客户一个运行时，或者采用一个可接受多个客户身份提供方的前端入口，并为每个客户配备独立的回调客户端。
基于密钥的客户端身份验证（private_key_jwt 或双向 TLS）替代客户端密钥。
使用 AWS PrivateLink 实现私有连接，并在 VPC 内部使用 TLS 实现端到端加密。
为 Amazon Simple Storage Service (Amazon S3)、Amazon ECR、AWS Secrets Manager、Amazon CloudWatch Logs 和 Amazon Bedrock Guardrails 使用客户托管的 AWS Key Management Service (AWS KMS) 密钥。
VPC Flow Logs 和 Network Load Balancer 访问日志。两个 API Gateway 阶段已默认写入访问日志。
为 Cognito 客户端凭证配置一个 Secrets Manager 轮换函数。
使用经过身份验证的入驻门户或 API 替代文件。

如果通过 API Gateway 的调用间歇性返回 HTTP 503，请检查内部 Network Load Balancer 是否开启了跨可用区负载均衡。每个 ECS 服务运行一个可启动于任一可用区的单一任务。

该仓库的 README 在 Security controls 和 Considerations for production 部分涵盖了完整列表。

安全注意事项

请仅在不含客户或业务数据的沙盒账户中部署此解决方案，并在测试后删除堆栈。该示例仅使用合成数据，不含个人数据。其合规智能体检查的是示例内部政策，不会评估或认证任何法律、法规或标准的合规性。该示例包含以下控制措施。在生产环境中使用此模式之前，你需要负责审查并调整它们：

传输加密。每个跨账户调用都使用 HTTPS 访问 Amazon API Gateway 或 Amazon Bedrock AgentCore，除上下文智能体的公共 Agent Card 外，每条路由都需要 Cognito JWT。
防护栏。在此示例中，每次模型调用都使用带提示攻击和有害内容过滤器的 Amazon Bedrock 防护栏，在模型使用来自另一组织的内容之前，会先对照同一防护栏进行检查。
最小权限 IAM。每个工作负载都有自己的角色，并指明确切的 ARN。只有编排器和合规智能体可以读取密钥，且各自只能读取自己的密钥。
容器加固。ECS 容器以只读根文件系统运行，并丢弃所有 Linux 能力。基础镜像通过摘要锁定。
账户隔离。两个账户不共享任何密钥、键或 IAM 权限。它们只交换公开值。

清理资源

为避免持续产生费用，请在完成测试后删除所有资源。运行第 3 个 notebook 底部的 Clean up 部分，然后是第 2 个 notebook，一次运行一个单元格。每部分的最后一个单元格会确认没有遗留内容。清理后，请检查两个账户中的 AWS CloudFormation、Amazon ECR 和 Amazon Cognito 控制台，以及 SaaS 提供商账户中的 Amazon Bedrock AgentCore 控制台，确认没有资源残留。

结论

本文展示了如何构建一个多智能体系统，让来自不同组织、基于不同框架构建的智能体通过 A2A 在 AWS 上进行协作。

Amazon ECS 上的 LangChain 编排器与 Amazon Bedrock AgentCore 上的 Strands 智能体通信。双方都不需要对方的框架或 SDK。SaaS 智能体回调企业上下文智能体以获取账户元数据，且双方各自使用自己的身份提供方进行身份验证。来自同一 ISV 的新技能无需修改代码。添加第二个 ISV 只需一次扩展，因为编排器目前只持有一个 OAuth 客户端：按照 README 中的 Considerations for production 所述，为每个 ISV 提供一个客户端和范围，然后运行相同的入驻交换流程，添加其 Agent Card URL，并重启编排器。

要开始使用，请克隆 GitHub 仓库，在两个账户中，使用你计算机上或 Amazon SageMaker AI notebook 实例上的 Jupyter 运行这三个 notebook。它们负责基础设施预置、智能体部署和端到端测试。

要了解更多关于本文中涉及的服务和协议，请参阅以下资源：

如果你有任何问题或反馈，请在评论区留言。