Python Agent 到 A2A Agent

connection 生成器可以将您的 Python Agent 连接到远程 A2A 代理 — 可以是 TypeScript 或 Python — 这样您的代理就可以将任务委托给另一个代理作为工具。

该生成器会设置所有必要的配置,使您的代理能够发现并调用远程 A2A 代理,无论是部署到 AWS(通过 Bedrock AgentCore)还是在本地运行时。

前置条件

在使用此生成器之前,请确保您具备:

1. 一个包含 Python Agent 组件的 Python 项目(Strands 或 LangChain)
2. 一个使用 --protocol=A2A 和 --auth=IAM 生成的 Agent 组件项目(可以是 ts#agent 或 py#agent)
3. 两个组件都使用 infra: agentcore 创建

使用方法

运行生成器

VSCode

1. 安装 Nx Console VSCode Plugin 如果您尚未安装
2. 在VSCode中打开Nx控制台
3. 点击 Generate (UI) 在"Common Nx Commands"部分
4. 搜索 @aws/nx-plugin - connection
5. 填写必需参数
6. 点击 Generate

CLI

pnpm

pnpm nx g @aws/nx-plugin:connection

yarn

yarn nx g @aws/nx-plugin:connection

npm

npx nx g @aws/nx-plugin:connection

bun

bunx nx g @aws/nx-plugin:connection

您还可以执行试运行以查看哪些文件会被更改

pnpm

pnpm nx g @aws/nx-plugin:connection --dry-run

yarn

yarn nx g @aws/nx-plugin:connection --dry-run

npm

npx nx g @aws/nx-plugin:connection --dry-run

bun

bunx nx g @aws/nx-plugin:connection --dry-run

选择您的宿主代理项目作为源,选择您的 A2A 代理项目作为目标。如果您的项目包含多个组件,请指定 sourceComponent 和 targetComponent 选项以消除歧义。

选项

参数	类型	默认值	描述
sourceProject 必需	string	-	源项目
targetProject 必需	string	-	要连接到的目标项目
sourceComponent	string	-	要从其连接的源组件（组件名称、相对于源项目根目录的路径或生成器 ID）。使用 '.' 显式选择项目作为源。
targetComponent	string	-	要连接到的目标组件（组件名称、相对于目标项目根目录的路径或生成器 ID）。使用 '.' 显式选择项目作为目标。
preferInstallDependencies	boolean	true	是否在生成器运行后优先安装依赖项。设置为 false 可在批量运行多个生成器时延迟安装（如果后续生成器需要计算 Nx 项目图，仍会运行安装）；在最后统一安装一次。

生成器输出

生成器会在 packages/common/agent_connection/ 创建一个共享的 agent_connection Python 项目(如果不存在)。每个连接的客户端模块都会生成到这个共享项目中:

• 文件夹packages/common/agent_connection

  • 文件夹<scope>_agent_connection

    • __init__.py 重新导出每个连接的客户端
    • 文件夹core

      • agentcore_endpoints.py 框架无关的 ARN/URL 解析
      • agentcore_a2a_client_config.py 框架无关的 A2A 客户端配置(签名的 ClientConfig)
      • agentcore_a2a_client_<framework>.py 为您的代理框架包装配置的 A2A 客户端
      • 文件夹auth/ 框架无关的 SigV4 / 会话转发 httpx.Auth

        • …
    • 文件夹app

      • <target_agent_name>_client_<framework>.py 每个 A2A 代理的连接 A2A 客户端

客户端后缀与您的代理框架匹配(_strands 或 _langchain)。两者都包装相同的框架无关签名 ClientConfig:Strands 客户端包装一个 Strands A2AAgent,而 LangChain 客户端直接驱动 a2a SDK。

此外,生成器还会:

• 转换您的代理的 agent.py,使用 @tool 将远程 A2A 代理注册为工具
• 将 agent_connection 项目添加为您的代理项目的工作区依赖项
• 更新代理的 serve-local 目标,使其依赖于目标代理的 serve-local 目标

使用连接的 A2A 代理

生成器会转换您的代理的 agent.py,将远程 A2A 代理包装为工具。远程代理通过 @tool 装饰的委托进行注册 — 装饰器的导入和代理构造函数因框架而异:

Strands

packages/my-project/my_module/agent/agent.py

from contextlib import contextmanager
from strands import Agent, tool

from my_scope_agent_connection import RemoteAgentClientStrands

@contextmanager
def get_agent():
    remote_agent = RemoteAgentClientStrands.create()

    @tool
    def ask_remote_agent(prompt: str) -> str:
        """Delegate a question to the remote RemoteAgent A2A agent and return its reply."""
        return str(remote_agent(prompt))

    yield Agent(
        system_prompt="...",
        tools=[ask_remote_agent],
    )

LangChain

packages/my-project/my_module/agent/agent.py

from langchain.agents import create_agent
from langchain_aws import ChatBedrockConverse
from langchain_core.tools import tool

from my_scope_agent_connection import RemoteAgentClientLangChain

def get_agent():
    remote_agent = RemoteAgentClientLangChain.create()

    @tool
    def ask_remote_agent(prompt: str) -> str:
        """Delegate a question to the remote RemoteAgent A2A agent and return its reply."""
        return str(remote_agent(prompt))

    return create_agent(
        model=ChatBedrockConverse(model=MODEL_ID, region_name=REGION),
        system_prompt="...",
        tools=[ask_remote_agent],
    )

两个客户端都可以直接调用,返回远程代理的回复,并包装一个 httpx.AsyncClient,该客户端在部署到 AWS 时使用 SigV4 签名请求,在 SERVE_LOCAL=true 时使用普通的 http://localhost:<port>/ 端点。Strands 客户端包装一个 Strands A2AAgent;LangChain 客户端直接驱动 a2a SDK,因此不会引入 Strands 依赖。

AgentCore 会话 ID 会通过 X-Amzn-Bedrock-AgentCore-Runtime-Session-Id 标头自动传播到远程代理,无论框架如何:代理服务器将入站请求的会话绑定到异步上下文中,连接客户端的签名 httpx.Auth 在每次出站调用时都会标记它 — 确保 Bedrock AgentCore Observability 的一致性。

基础设施

运行连接生成器后,您需要授予宿主代理调用远程 A2A 代理的权限。

CDK

packages/infra/src/stacks/application-stack.ts

const remoteAgent = new RemoteAgent(this, 'RemoteAgent');
const myAgent = new MyAgent(this, 'MyAgent');

// Grant the host agent permission to invoke the remote A2A agent
remoteAgent.grantInvokeAccess(myAgent);

A2A 代理上的 grantInvokeAccess 会同时配置 bedrock-agentcore:InvokeAgentRuntime 和 bedrock-agentcore:GetAgentCard — A2A 客户端需要这两者来获取代理卡片并发送消息。

远程代理的 AgentCore 运行时 ARN 会由生成的 CDK 构造自动注册到运行时配置的 agentcore 命名空间中,以便宿主代理可以在运行时发现它。

Terraform

packages/infra/src/main.tf

module "remote_agent" {
  source = "../../common/terraform/src/app/agents/remote-agent"

  appconfig_application_id  = module.runtime_config_appconfig.application_id
  appconfig_application_arn = module.runtime_config_appconfig.application_arn
}

module "my_agent" {
  source = "../../common/terraform/src/app/agents/my-agent"

  appconfig_application_id  = module.runtime_config_appconfig.application_id
  appconfig_application_arn = module.runtime_config_appconfig.application_arn
}

# Grant the host agent permission to invoke the remote A2A agent
resource "aws_iam_policy" "agent_invoke_a2a" {
  name = "AgentInvokeA2aPolicy"
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect = "Allow"
      Action = [
        "bedrock-agentcore:InvokeAgentRuntime",
        "bedrock-agentcore:GetAgentCard",
      ]
      Resource = module.remote_agent.agent_core_runtime_arn
    }]
  })
}

resource "aws_iam_role_policy_attachment" "agent_invoke_a2a" {
  role       = module.my_agent.agent_core_runtime_role_arn
  policy_arn = aws_iam_policy.agent_invoke_a2a.arn
}

远程代理的 AgentCore 运行时 ARN 会由生成的 Terraform 模块自动注册到运行时配置的 agentcore 命名空间中,以便宿主代理可以在运行时发现它。

本地开发

生成器配置宿主代理的 dev 目标以:

1. 自动启动连接的 A2A 代理
2. 设置 LOCAL_DEV=true,使生成的客户端直接连接到 http://localhost:<port>/ 而不是 AgentCore

使用以下命令在本地运行代理:

pnpm

pnpm nx <agent-name>-dev <project-name>

yarn

yarn nx <agent-name>-dev <project-name>

npm

npx nx <agent-name>-dev <project-name>

bun

bunx nx <agent-name>-dev <project-name>

这将启动宿主代理和所有连接的 A2A 代理,宿主代理通过分配的本地端口使用普通 HTTP 调用远程代理。 程代理。