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

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

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

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

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

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

在《智能体互操作性开放协议第 4 部分：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 链接和内部 Network Load Balancer 私密地将请求转发给 Enterprise Orchestrator，这是一个 LangChain agent。
当其模型判断该问题需要进行合规扫描时，orchestrator 会使用存储在 AWS Secrets Manager 中的客户端密钥，从企业用户池中获取自己的令牌。
orchestrator 使用它在启动时从 agent 的 Agent Card 中读取的技能，向运行在 Amazon Bedrock AgentCore 运行时上的合规 SaaS agent 发送 A2A 请求。
合规 agent 在入职时注册的回调 URL 处检查企业上下文 agent 的 Agent Card，然后从 SaaS 提供商自己的池中获取令牌。
合规 agent 回调企业上下文 agent 以获取账户元数据，完成扫描，并将答案返回给 orchestrator，然后再返回给应用。

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

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

两个企业 agent 都运行在 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 等服务。对于此演示，agent 使用模拟跨多个账户合规发现的合成数据。

在扫描之前，SaaS 代理会回调 Enterprise Context Agent 以获取企业上下文，例如哪些账户是生产账户、适用哪些内部策略，以及应向谁上报。这种双向模式使各代理成为真正的对等方，而不是客户端与服务器的关系。回调只会发往客户在入门引导时注册的地址。

关键设计决策是使用 A2A 进行所有跨边界通信。企业无法访问 SaaS 代理的内部工具、系统提示或模型选择。这种隔离意味着，只要其 Agent Card 和 A2A 接口保持兼容，ISV 就可以更改其内部架构而无需更改客户集成。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 IAM Identity Center 配置文件，并使用 aws configure sso 进行设置。你不需要 Docker：AWS CodeBuild 会构建容器镜像。
- 或者，在每个账户中有一个 Amazon SageMaker AI notebook 实例（ml.t3.medium 就足够了）。请参阅 Amazon SageMaker AI 设置指南。
在每个账户中，有一个 AWS Identity and Access Management (IAM) 角色，具有创建 AWS CloudFormation 堆栈及其中的资源的权限。每个 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 的定价页面。完成后请运行“清理”部分。

实施解决方案

使用三个笔记本部署系统，在 Jupyter 中按顺序逐个单元格运行。笔记本 1（企业账户，约 3 分钟）创建企业身份并写入入职请求。笔记本 2（SaaS 提供商账户，约 5 分钟）部署合规代理并写入入职响应。笔记本 3（企业账户，约 20 分钟）部署企业代理并运行测试。入职文件仅包含公开值。

克隆 GitHub 仓库，并为本地运行设置两个 AWS 配置文件，或者为每个账户设置一个笔记本实例，并将仓库克隆到每个实例。
在企业账户中运行笔记本 1。
将 onboarding-request.json 交给 SaaS 提供商。
在 SaaS 提供商账户中运行笔记本 2。
将 onboarding-response.json 交回企业。
在企业账户中运行笔记本 3。
测试完成后，先运行笔记本 3 的“清理”，再运行笔记本 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 agent 通过 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)
            ...

每一次失败，从卡片未验证到令牌被拒绝，都只给模型一条消息：“Enterprise context unavailable”，而详细信息保留在 SaaS 提供商的日志中。

在 AgentCore 上通过 A2A 暴露 agent

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 文档，描述 agent 的名称、能力和技能。兼容 A2A 的客户端可以获取此卡片，无需文档或 SDK 集成即可了解该 agent 能做什么。

使用 LangChain 构建企业编排器

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

编排器从企业自己的 Cognito 池获取令牌，并仅通过 HTTPS 将 JSON-RPC 负载发送到固定为获取 Agent Card 地址的 agent URL。这段精简代码展示了核心 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 发现 agent 并路由请求

启动时，编排器从配置的 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 会随之变化，编排器在下次重启时会获取新技能。两个团队无需协调变更。

构建企业上下文 agent

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

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

使用 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 镜像，但 CloudFormation 创建的是存放这些镜像的 Amazon Elastic Container Registry (Amazon ECR) 存储库。您可以在 AWS CloudFormation 控制台中监控这些堆栈。

笔记本会尽早暴露失败。每个笔记本首先通过一次简短测试调用检查 Amazon Bedrock 模型访问权限，然后才部署任何内容，并且每一步都会在第一个错误处停止并给出清晰的输出。清理部分位于笔记本 2 和 3 的底部，在一个停止单元格之后，因此“全部运行”永远不会到达它们。

了解 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"])

编排器通过 SaaS 代理的 Agent Card 发现它，并通过 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 底部的“清理”部分，然后运行笔记本 2，每次运行一个单元格。每个部分的最后一个单元格会确认没有遗留任何内容。清理后，请检查两个账户中的 AWS CloudFormation、Amazon ECR 和 Amazon Cognito 控制台，以及 SaaS 提供商账户中的 Amazon Bedrock AgentCore 控制台，以确认没有遗留任何资源。

结论

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

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

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

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

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

关于作者

Amit Chaudhary

Amit 是 Amazon Web Services 的高级解决方案架构师，他与 ISV 和企业客户合作，设计并构建可在生产环境中运行的生成式 AI 解决方案。他专注于多智能体架构、大型语言模型以及 AWS 上的 AI 应用开发。