Retrofit, don’t rebuild: Agentic overlays for transforming legacy enterprise services

"""
A2A 请求转换器 - 计算器示例。

此模块实现请求转换器模式，以在现有的计算器 REST API 上提供 A2A
(JSON-RPC 2.0) 兼容性。

A2A 规范 0.3 兼容性：
- 代理卡片：GET /.well-known/agent-card.json
- JSON-RPC 端点：POST /a2a
- 方法：SendMessage, SendStreamingMessage
- 消息格式：{ "message": { "parts": [{ "kind": "data", "data": {...} }] } }
"""

# 默认 A2A API URL（如需覆盖，请使用 build_agent_card(url)）
A2A_API_URL = "http://localhost:5000/a2a"

EXECUTE_TIMEOUT_SECONDS = 30

# 从 JSON 文件加载技能
_SKILLS_FILE = Path(__file__).parent / "skills.json"
_SKILLS_CACHE: Optional[List[Dict[str, Any]]] = None

def _load_skills() -> List[Dict[str, Any]]:
    """
    从 skills.json 文件加载技能。
    首次加载后，技能会被缓存以避免重复读取文件。
    """
    global _SKILLS_CACHE
    if _SKILLS_CACHE is not None:
        return _SKILLS_CACHE
    try:
        with open(_SKILLS_FILE) as f:
            _SKILLS_CACHE = json.load(f)
            return _SKILLS_CACHE
    except FileNotFoundError:
        logger.error(f"未找到技能文件: {_SKILLS_FILE}")
        return []
    except json.JSONDecodeError as e:
        logger.error(f"技能文件中的 JSON 无效: {e}")
        return []

def build_agent_card(api_url: Optional[str] = None) -> Dict[str, Any]:
    """构建 A2A Agent Card（v0.3.0 格式），并支持可配置的 API URL。"""
    if api_url is None:
        api_url = A2A_API_URL
    return {
        "name": "Calculator Agent",
        "description": "支持基本算术运算的简单计算器",
        "supportedInterfaces": [
            {"url": api_url, "protocolBinding": "JSONRPC", "protocolVersion": "0.3"},
        ],
        "provider": {
            "organization": "Example Organization",
            "url": "",
        },
        "version": "1.0.0",
        "capabilities": {
            "streaming": False,
            "pushNotifications": False,
            "extendedAgentCard": False,
        },
        "defaultInputModes": ["text/plain", "application/json"],
        "defaultOutputModes": ["text/plain", "application/json"],
        "skills": _load_skills(),
    }

# 动态构建 Agent Card
AGENT_CARD = build_agent_card()

def invoke_rest_endpoint(
    endpoint: str,
    json_data: Optional[Dict] = None,
    http_method: str = "POST"
) -> Tuple[Optional[Dict], int]:
    """
    通过真实的 HTTP 请求调用内部 REST 端点。

    使用 requests.post/get 调用正在运行的服务器。这确保了
    中间件、装饰器和标头都能被正确测试。

    参数:
        endpoint: REST 端点路径（例如 "/api/v1/calculate"）
        json_data: POST/PUT 请求体
        http_method: HTTP 方法（GET、POST 等）

    返回:
        元组（response_data, status_code）
    """
    try:
        base_url = request.host_url.rstrip("/")
        url = f"{base_url}{endpoint}"

        headers = {"Content-Type": "application/json"}
        auth_header = request.headers.get("Authorization")
        if auth_header:
            headers["Authorization"] = auth_header

        logger.info(f"适配器: 代理到 REST {http_method} {url}")

        if http_method.upper() == "POST":
            response = http_requests.post(
                url, json=json_data, headers=headers,
                timeout=EXECUTE_TIMEOUT_SECONDS
            )
        elif http_method.upper() == "GET":
            response = http_requests.get(
                url, headers=headers,
                timeout=EXECUTE_TIMEOUT_SECONDS
            )
        else:
            response = http_requests.request(
                http_method, url, json=json_data, headers=headers,
                timeout=EXECUTE_TIMEOUT_SECONDS
            )

        logger.info(f"适配器: REST 返回 {response.status_code}")
        return response.json(), response.status_code

    except http_requests.RequestException as e:
        logger.error(f"适配器: 调用 REST 端点时出错: {e}", exc_info=True)
        return {"error": "内部服务器错误"}, 500

def extract_message_payload(message: Dict) -> Optional[Dict]:
    """
    从 A2A 消息部分中提取负载（规范 0.3 格式）。

    预期格式：
    {
        "message": {
            "parts": [{"kind": "data", "data": {"operation": "add", "operands": [5, 3]}}]
        }
    }

    返回：
        提取的负载数据作为字典，如果未找到则返回 None
    """
    try:
        parts = message.get("parts", [])
        for part in parts:
            if isinstance(part, dict) and part.get("kind") == "data":
                return part.get("data")
        return None
    except Exception as e:
        logger.error(f"提取消息负载时出错：{e}")
        return None

def build_a2a_message(message_id: str, context_id: str, content: Any) -> Dict:
    """
    构建符合 A2A 规范的消息对象（A2A 规范 0.3）。

    响应格式：
    {
        "messageId": "uuid",
        "contextId": "uuid",
        "role": "agent",
        "parts": [{"kind": "data", "data": {...}}],
        "kind": "message",
        "metadata": {}
    }
    """
    if isinstance(content, dict):
        parts = [{"kind": "data", "data": content}]
    else:
        parts = [{"kind": "text", "text": str(content)}]

    return {
        "messageId": message_id,
        "contextId": context_id,
        "role": "agent",
        "parts": parts,
        "kind": "message",
        "metadata": {}
    }

# 根据 JSON-RPC 2.0 规范定义的错误代码
class JsonRpcError:
    PARSE_ERROR = -32700
    INVALID_REQUEST = -32600
    METHOD_NOT_FOUND = -32601
    INVALID_PARAMS = -32602
    INTERNAL_ERROR = -32603

def jsonrpc_error(code: int, message: str, data: Any = None, request_id: Any = None) -> Dict:
    """构建 JSON-RPC 2.0 错误响应。"""
    response = {
        "jsonrpc": "2.0",
        "error": {"code": code, "message": message},
        "id": request_id
    }
    if data is not None:
        response["error"]["data"] = data
    return response

def jsonrpc_success(result: Any, request_id: Any = None) -> Dict:
    """构建 JSON-RPC 2.0 成功响应。"""
    return {
        "jsonrpc": "2.0",
        "result": result,
        "id": request_id
    }

def handle_send_message(data: Dict) -> Tuple[Any, int]:
    """
    处理 SendMessage —— 直接传递到 /api/v1/calculate。
    适配器不会检查或根据负载内容进行路由。
    """
    request_id = data.get("id")
    params = data.get("params", {})
    message = params.get("message", {})
    context_id = message.get("contextId") or generate_id()
    message_id = generate_id()

    payload = extract_message_payload(message)
    if not payload:
        return jsonify(jsonrpc_error(
            JsonRpcError.INVALID_PARAMS,
            "无效参数：消息中未找到 message.parts 的数据。",
            request_id=request_id
        )), 400

### 第 6 步：设置 A2A 路由（规范 0.3）

官方的 A2A SDK 提供了适用于 FastAPI 和 Starlette 应用的 A2A 库，这些库可以抽象出添加 A2A 特定路由的复杂性。虽然 A2A 库也适用于 Flask 应用，但我们在示例代码中没有使用它们。我们希望让您可以轻松理解如何在 Flask 应用上托管 A2A 覆盖层。以下代码片段添加了 A2A 所需的路由。

### 第 7 步：最后，在您的应用中进行初始化

### 第 8 步：运行您的应用

从项目根目录运行以下命令。

## 使用 Amazon Bedrock AgentCore Gateway 添加代理覆盖层

Amazon Bedrock AgentCore 是一项服务，用于在无需管理基础设施的情况下构建、连接和优化代理。如以下图表所示，AgentCore Gateway 可以通过作为端点和服务的单一接入点，将代理覆盖层与应用解耦。这种分离使得一个代理覆盖层可以服务于多个服务或应用，而不仅仅是一个。AgentCore Gateway 每个网关最多支持 10 个目标，并且可以原生集成到现有的 AWS 服务中，并支持 OpenAPI 端点。

图 5：使用 Amazon Bedrock AgentCore Gateway 解耦代理层

企业级应用程序通常需要协调多个服务来处理复杂任务。例如，一个计算器应用程序处理“计算 2 * (3 + 4)”。如以下图表所示，系统首先查询一个运算顺序端点（如 /api/order-of-ops/...）以确定评估顺序。然后，它依次调用一个端点（如 /api/arithmetic/...）来计算“3 + 4”，接着是“2 * 7”。如果为每个服务添加代理层，会引入额外的开销。相反，您可以将这两个服务整合到一个代理层中，根据需要协调调用。

您可以将代理层与应用程序分离，以便根据功能而非仅根据应用程序来组织代理层。代理层作为代理方式与您的应用程序和系统进行交互，作为代理方式与功能进行交互。

除了 AgentCore Gateway 之外，AgentCore 的其他功能还可以简化代理层的监控、迭代和部署。AgentCore Identity 处理代理和网关组件的认证，支持 OAuth 2.0 提供商，并提供对 Okta、GitHub 和 Slack 的托管集成。AgentCore Observability 通过指标、日志和跨度可视化监控代理性能。您可以查看工具调用和延迟等高层数据，或使用与 Amazon CloudWatch 的原生集成，检查跨组件的详细执行路径。AgentCore Runtime 通过容器镜像部署模型，无论是开源模型、自定义模型，还是 Amazon Bedrock LLM（如 Nova 和 Anthropic Claude），而无需您管理大型语言模型（LLM）基础设施。

图 6：使用 Amazon Bedrock AgentCore 功能解耦代理层 —— 运行时、网关、身份和可观测性

通过使用 AgentCore 的不同功能，您可以简化代理和代理层的部署，并轻松集成到现有的 AWS 堆栈中。由于它是一个托管服务，它还可以减少实现代理层所需的工作量。

## 与合作伙伴共同加速企业 AI 采用

这种代理层模式代表了作者与 AWS 之间更广泛的协作，以帮助企业在现有基础设施和新兴 AI 能力之间架起桥梁。成功的 AI 采用需要务实的解决方案，既要尊重现有投资，又要支持渐进式转型。AWS 和作者正在共同开发参考架构、实现模式和工具，使企业能够在不进行大规模基础设施更换的情况下采用 A2A 通信。代理层模式体现了这一理念：保留已有的有效部分，在需要时进行扩展，并提供清晰的迁移路径，以最小化风险并最大化价值。

## 结论

代理层为企业提供了一条务实的路径，使其能够采用代理到代理（Agent-to-Agent）通信，而无需放弃其现有的 REST API 投资。通过添加一个轻量的翻译层，将 A2A 消息转换为 REST 负载，组织可以在新兴的代理生态系统中参与进来，同时保留稳定且经过生产测试的业务逻辑。两种实现模式提供了灵活性，以满足组织的不同需求：适用于特定使用场景的内部应用代理层，以及适用于企业级部署的 AgentCore 网关。无论您是启用一个单一的监督代理，还是协调复杂的多服务工作流，代理层都能减少并行基础设施的操作开销，以及全面重构所带来的回归风险。

当企业从确定性的 REST API 过渡到基于推理的代理系统时，关键的洞察始终是：A2A 并不是一个全新的 API，而是对现有 API 的一种新接口。这种观点将采用的挑战从重新构建一切转变为逐步适配，使组织能够更快地实现 AI 价值，同时有效管理风险。对于准备探索代理层的组织，计算器示例提供了一个具体的起点，而 AgentCore 则为生产部署提供了基础设施。

## 下一步

- 评估您的架构 – 审计 REST 服务，寻找适合 A2A 启用的候选服务。内部应用代理层适用于单一服务代理。AgentCore 网关适用于多服务工作流。

- 查看参考实现 – Flask 计算器示例展示了翻译模式，包括代理卡片的设置、消息提取、REST 调用和响应构建。

- 探索 Amazon Bedrock AgentCore – AgentCore 网关、身份验证和可观测性为生产级代理层提供了基础设施。

- 加入 A2A 社区 – A2A 协议规范和 SDK 文档可在 [a2a-protocol.org](https://a2a-protocol.org) 获取，支持 Flask、FastAPI 和 Starlette 应用的库也一并提供。

企业 AI 的未来并不在于替换现有系统，而在于通过代理能力来扩展它们。代理层使这一未来今天即可实现。

## 作者简介

'"`