A2A 中的扩展¶

Agent2Agent (A2A) 协议为代理间通信提供了坚实的基础。然而，特定领域或高级用例通常需要额外的结构、自定义数据或超出通用方法的新交互模式。扩展是 A2A 强大的机制，用于在基本协议之上叠加新功能。

扩展允许通过新的数据、要求、RPC 方法和状态机来扩展 A2A 协议。代理在其 Agent Card 中声明对特定扩展的支持，然后客户端可以在向代理发出的请求中选择启用扩展提供的行为。扩展由 URI 标识并由其自己的规范定义。任何人都可以定义、发布和实现扩展。

扩展的灵活性允许自定义 A2A，而不会碎片化核心标准，从而促进创新和特定领域的优化。

扩展的范围¶

使用扩展的精确方式是故意宽泛的，以便能够将 A2A 扩展到已知用例之外。然而，一些可预见的应用程序包括

仅数据扩展：在 Agent Card 中公开新的、结构化的信息，而不影响请求-响应流。例如，一个扩展可以添加有关代理 GDPR 合规性的结构化数据。
配置文件扩展：在核心请求-响应消息上叠加额外的结构和状态更改要求。此类型有效地充当核心 A2A 协议的配置文件，缩小了允许值的范围（例如，要求所有消息都使用符合特定模式的 DataParts）。这还可以通过使用元数据来增强任务状态机中的现有状态。例如，当 TaskStatus.state 为“working”且 TaskStatus.message.metadata["generating-image"] 为 true 时，扩展可以定义一个“generating-image”子状态。
方法扩展（扩展技能）：添加超出协议定义的核心集的全新 RPC 方法。扩展技能是指代理通过实现定义新 RPC 方法的扩展而获得或公开的能力或功能。例如，task-history 扩展可能会添加一个 tasks/search RPC 方法来检索以前任务的列表，从而有效地为代理提供一项新的扩展技能。
状态机扩展：向任务状态机添加新状态或转换。

示例扩展列表¶

扩展	描述
安全护照扩展	添加受信任的上下文层，用于即时个性化和减少开销 (v1)。
Hello World 或时间戳扩展	一个简单的扩展，演示如何通过向 `Message` 和 `Artifact` 对象的 `metadata` 字段添加时间戳来增强基础 A2A 类型 (v1)。
可追溯性扩展	探索可追溯性扩展的 Python 实现和基本用法 (v1)。
代理网关协议 (AGP) 扩展	一个核心协议层或路由扩展，引入了自治小队 (ASq) 并根据声明的能力路由意图负载，从而增强了可伸缩性 (v1)。

限制¶

有些协议更改是扩展不允许的，主要是为了防止破坏核心类型验证

更改核心数据结构的定义：例如，向协议定义的数据结构添加新字段或删除必需字段）。扩展应将自定义属性放置在核心数据结构上存在的 metadata 映射中。
向枚举类型添加新值：扩展应使用现有的枚举值并在 metadata 字段中注释附加的语义含义。

扩展声明¶

代理通过在其 AgentCapabilities 对象中包含 AgentExtension 对象来在其 Agent Card 中声明对扩展的支持。

以下是带扩展的 Agent Card 示例

{
  "name": "Magic 8-ball",
  "description": "An agent that can tell your future... maybe.",
  "version": "0.1.0",
  "url": "https://example.com/agents/eightball",
  "capabilities": {
    "streaming": true,
    "extensions": [
      {
        "uri": "https://example.com/ext/konami-code/v1",
        "description": "Provide cheat codes to unlock new fortunes",
        "required": false,
        "params": {
          "hints": [
            "When your sims need extra cash fast",
            "You might deny it, but we've seen the evidence of those cows."
          ]
        }
      }
    ]
  },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain"],
  "skills": [
    {
      "id": "fortune",
      "name": "Fortune teller",
      "description": "Seek advice from the mystical magic 8-ball",
      "tags": ["mystical", "untrustworthy"]
    }
  ]
}

必需的扩展¶

虽然扩展通常提供可选功能，但某些代理可能具有更严格的要求。当 Agent Card 将扩展声明为 required: true 时，它向客户端发出信号，表明扩展的某些方面会影响请求的结构或处理方式，并且客户端必须遵守。代理不应将仅数据扩展标记为必需。如果客户端未请求激活必需的扩展，或未能遵循其协议，则代理应拒绝传入请求并返回适当的错误。

扩展规范¶

扩展的详细行为和结构由其规范定义。虽然没有强制要求确切的格式，但它至少应包含

标识扩展的特定 URI。
AgentExtension 对象的 params 字段中指定的对象的模式和含义。
客户端和代理之间通信的任何附加数据结构的模式。
新请求-响应流、附加端点或实现扩展所需的任何其他逻辑的详细信息。

扩展依赖项¶

扩展可能依赖于其他扩展。这可以是必需的依赖项（扩展在没有依赖项的情况下无法运行）或可选的依赖项（如果存在另一个扩展，则启用附加功能）。扩展规范应记录这些依赖项。客户端有责任激活扩展及其规范中列出的所有必需依赖项。

扩展激活¶

扩展默认处于非活动状态，为不了解扩展的客户端提供基线体验。客户端和代理执行协商以确定哪些扩展在特定请求中处于活动状态。

客户端请求：客户端通过在向代理发出的 HTTP 请求中包含 A2A-Extensions 标头来请求扩展激活。该值是客户端打算激活的扩展 URI 的逗号分隔列表。
代理处理：代理负责识别请求中支持的扩展并执行激活。任何请求但代理不支持的扩展都可以忽略。
响应：一旦代理识别出所有已激活的扩展，响应 SHOULD 包含 A2A-Extensions 标头，列出该请求成功激活的所有扩展。

A2A Extension Flow Diagram

显示扩展激活的示例请求

POST /agents/eightball HTTP/1.1
Host: example.com
Content-Type: application/json
A2A-Extensions: https://example.com/ext/konami-code/v1
Content-Length: 519
{
  "jsonrpc": "2.0",
  "method": "message/send",
  "id": "1",
  "params": {
    "message": {
      "kind": "message",
      "messageId": "1",
      "role": "user",
      "parts": [{"kind": "text", "text": "Oh magic 8-ball, will it rain today?"}]
    },
    "metadata": {
      "https://example.com/ext/konami-code/v1/code": "motherlode"
    }
  }
}

回显已激活扩展的相应响应

HTTP/1.1 200 OK
Content-Type: application/json
A2A-Extensions: https://example.com/ext/konami-code/v1
Content-Length: 338
{
  "jsonrpc": "2.0",
  "id": "1",
  "result": {
    "kind": "message",
    "messageId": "2",
    "role": "agent",
    "parts": [{"kind": "text", "text": "That's a bingo!"}]
  }
}

实施注意事项¶

虽然 A2A 协议定义了扩展的功能，但本节提供了有关其实现的指南——编写、版本控制和分发扩展实现的最佳实践。

版本控制：扩展规范不断发展。拥有清晰的版本控制策略至关重要，以确保客户端和代理可以协商兼容的实现。
- 建议：使用扩展的 URI 作为主要版本标识符，最好包含版本号（例如，https://example.com/ext/my-extension/v1）。
- 破坏性更改：当对扩展的逻辑、数据结构或必需参数进行破坏性更改时，必须使用新的 URI。
- 处理不匹配：如果客户端请求代理不支持的版本，代理 SHOULD 忽略该扩展的激活请求；它 MUST NOT 回退到不同的版本。
可发现性和发布:
- 规范托管：扩展规范文档应该托管在扩展的 URI 上。
- 永久标识符：鼓励作者使用永久标识符服务（例如 w3id.org）作为其扩展 URI，以防止链接失效。
- 社区注册表（未来）：A2A 社区未来可能会建立一个中央注册表，用于发现和浏览可用的扩展。

打包和可重用性 (A2A SDK 和库)：为了促进采用，扩展逻辑应打包成可重用库，可以集成到现有的 A2A 客户端和服务器应用程序中。

扩展实现应作为其语言生态系统的标准包分发（例如，Python 的 PyPI 包，TypeScript/JavaScript 的 npm 包）。

目标是为开发人员提供简化的集成体验。设计良好的扩展包应允许开发人员以最少的代码将其添加到其服务器，例如

import logging
import os

import click

from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from agent import ReimbursementAgent
from agent_executor import ReimbursementAgentExecutor
from dotenv import load_dotenv
from timestamp_ext import TimestampExtension


load_dotenv()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class MissingAPIKeyError(Exception):
    """Exception for missing API key."""


@click.command()
@click.option('--host', default='localhost')
@click.option('--port', default=10002)
def main(host, port):
    try:
        # Check for API key only if Vertex AI is not configured
        if not os.getenv('GOOGLE_GENAI_USE_VERTEXAI') == 'TRUE':
            if not os.getenv('GEMINI_API_KEY'):
                raise MissingAPIKeyError(
                    'GEMINI_API_KEY environment variable not set and GOOGLE_GENAI_USE_VERTEXAI is not TRUE.'
                )

        hello_ext = TimestampExtension()
        capabilities = AgentCapabilities(
            streaming=True,
            extensions=[
                hello_ext.agent_extension(),
            ],
        )
        skill = AgentSkill(
            id='process_reimbursement',
            name='Process Reimbursement Tool',
            description='Helps with the reimbursement process for users given the amount and purpose of the reimbursement.',
            tags=['reimbursement'],
            examples=[
                'Can you reimburse me $20 for my lunch with the clients?'
            ],
        )
        agent_card = AgentCard(
            name='Reimbursement Agent',
            description='This agent handles the reimbursement process for the employees given the amount and purpose of the reimbursement.',
            url=f'http://{host}:{port}/',
            version='1.0.0',
            default_input_modes=ReimbursementAgent.SUPPORTED_CONTENT_TYPES,
            default_output_modes=ReimbursementAgent.SUPPORTED_CONTENT_TYPES,
            capabilities=capabilities,
            skills=[skill],
        )
        agent_executor = ReimbursementAgentExecutor()
        # Use the decorator version of the extension for highest ease of use.
        agent_executor = hello_ext.wrap_executor(agent_executor)
        request_handler = DefaultRequestHandler(
            agent_executor=agent_executor,
            task_store=InMemoryTaskStore(),
        )
        server = A2AStarletteApplication(
            agent_card=agent_card, http_handler=request_handler
        )
        import uvicorn

        uvicorn.run(server.build(), host=host, port=port)
    except MissingAPIKeyError as e:
        logger.error(f'Error: {e}')
        exit(1)
    except Exception as e:
        logger.error(f'An error occurred during server startup: {e}')
        exit(1)


if __name__ == '__main__':
    main()

此示例展示了 A2A SDK 或库（例如 Python 中的 a2a.server）如何促进 A2A 代理和扩展的实现。

安全性：扩展修改了 A2A 协议的核心行为，因此引入了新的安全考虑
- 输入验证：扩展引入的任何新数据字段、参数或方法都必须经过严格验证。将来自外部方的所有与扩展相关的数据视为不受信任的输入。
- 必需扩展的范围：在 Agent Card 中将扩展标记为 required: true 时要小心。这会为所有客户端创建硬依赖项，并且应仅用于对代理的核心功能和安全性至关重要的扩展（例如，消息签名扩展）。
- 身份验证和授权：如果扩展添加了新方法，则实现必须确保这些方法与核心 A2A 方法受到相同的身份验证和授权检查。扩展不得提供绕过代理主要安全控制的方法。

有关更多信息，请参阅 A2A Extensions: Empowering Custom Agent Functionality 博客文章。