A2A 协议扩展：安全护照 (Secure Passport) 完整指南

目录

1. 完整概述
2. 详细技术规范
3. 四个实际使用场景
4. 完整 Python 实现指南
5. 最佳实践

1. 完整概述

扩展目的和核心优势

安全护照扩展 (Secure Passport Extension) 为 Agent2Agent (A2A) 协议引入了一个可信的上下文层，使调用代理能够安全且自愿地与被调用代理共享其当前上下文状态的结构化子集。这个扩展旨在将匿名的、事务性的调用转换为协作的合作伙伴关系。

核心优势

• 即时个性化：专业代理可以立即使用上下文（如忠诚度等级或首选货币）
• 减少开销：消除了建立上下文所需的多轮对话
• 增强信任：包含**signature**字段，用于对数据来源和完整性进行加密验证

技术规范和标识符

• URI: https://github.com/a2aproject/a2a-samples/tree/main/samples/python/extensions/secure-passport
• 类型: 配置文件扩展 / 纯数据扩展
• 版本: 1.0.0
• 协议: Agent2Agent (A2A)

核心优势详解

即时个性化

专业代理可以立即访问调用者的上下文信息，无需额外的身份验证或数据收集步骤。例如，旅行代理可以立即知道用户的货币偏好和忠诚度等级。

减少开销

通过预共享上下文信息，避免了传统的"你是谁？你需要什么？"式的多轮对话，显著提高了代理间通信的效率。

增强信任

数字签名机制确保上下文数据的完整性和来源验证，为高权限操作提供安全保障。

2. 详细技术规范

完整的 CallerContext 数据结构

CallerContext 对象是安全护照的核心负载，包含以下字段：

消息集成示例（实际 JSON）

完整的 A2A 消息请求示例

{
  "jsonrpc": "2.0",
  "id": "req-123",
  "method": "tasks/send",
  "params": {
    "message": {
      "messageId": "msg-456",
      "role": "user",
      "parts": [
        {"kind": "text", "content": "Book a flight for me."}
      ],
      "metadata": {
        "https://github.com/a2aproject/a2a-samples/tree/main/samples/python/extensions/secure-passport": {
          "clientId": "a2a://orchestrator-agent.com",
          "sessionId": "travel-session-xyz",
          "signature": "MOCK-SIG-123456...",
          "state": {
            "user_preferred_currency": "GBP",
            "loyalty_tier": "Gold"
          }
        }
      }
    }
  }
}

CallerContext 负载示例

{
  "clientId": "a2a://orchestrator-agent.com",
  "sessionId": "travel-session-xyz",
  "signature": "MOCK-SIG-123456...",
  "state": {
    "user_preferred_currency": "GBP",
    "loyalty_tier": "Gold"
  }
}

代理声明和协商模式

AgentCard 声明示例

能够接收和利用安全护照上下文的 A2A 代理必须在其 AgentCard 的 AgentCapabilities 对象的 extensions 部分声明其支持。

{
  "uri": "https://github.com/a2aproject/a2a-samples/tree/main/samples/python/extensions/secure-passport",
  "params": {
    "supportedStateKeys": ["user_preferred_currency", "loyalty_tier"]
  }
}

被调用代理使用 supportedStateKeys 数组明确声明它理解并优化使用的上下文数据键。

3. 四个实际使用场景

场景 1：货币转换（高信任度）

场景描述：旅行编排器需要货币转换服务，需要传递用户的货币偏好。

上下文数据：

{
  "clientId": "a2a://travel-orchestrator.com",
  "state": {
    "user_preferred_currency": "GBP",
    "user_id": "U001"
  },
  "signature": "sig-currency-1"
}

优势：

• 货币转换代理可以立即以用户偏好的货币显示价格
• 数字签名确保货币偏好的可信度
• 无需额外的身份验证步骤

场景 2：个性化旅行预订（会话数据）

场景描述：旅行门户需要为高忠诚度用户提供个性化服务。

上下文数据：

{
  "clientId": "a2a://travel-portal.com",
  "sessionId": "travel-booking-session-999",
  "state": {
    "destination": "Bali, Indonesia",
    "loyalty_tier": "Platinum"
  },
  "signature": "sig-travel-2"
}

优势：

• 旅行代理可以立即识别高价值客户
• 会话 ID 支持跨多个请求的上下文连续性
• 可以根据忠诚度等级提供特殊优惠

场景 3：零售协助（无签名，低风险）

场景描述：电商前端需要产品推荐服务，风险较低，无需签名验证。

上下文数据：

{
  "clientId": "a2a://ecommerce-front.com",
  "state": {
    "product_sku": "Nikon-Z-50mm-f1.8",
    "user_intent": "seeking_reviews"
  }
}

优势：

• 推荐代理可以立即了解用户意图
• 无签名要求，适合低风险场景
• 快速响应，无需验证开销

场景 4：安全数据库访问（严格权限）

场景描述：营销代理需要访问财务数据库，需要严格的权限控制。

上下文数据：

{
  "clientId": "a2a://marketing-agent.com",
  "state": {
    "query_type": "quarterly_revenue",
    "access_scope": ["read:finance_db", "user:Gulli"]
  },
  "signature": "sig-finance-4"
}

优势：

• 数据库代理可以验证访问权限
• 数字签名确保请求的合法性
• 支持细粒度的权限控制

4. 完整 Python 实现指南

安装和设置说明

1. 环境要求

• Python 3.9 或更高版本
• Poetry（推荐用于依赖管理）

2. 项目设置

# 导航到示例项目目录
cd extensions/secure-passport/v1/samples/python

# 安装依赖
poetry install

# 激活虚拟环境
poetry shell

核心实现（代码示例）

1. 基本数据模型

from typing import Optional, Dict, Any
from pydantic import BaseModel, Field, ConfigDict

class CallerContext(BaseModel):
    """
    安全护照负载，包含调用代理共享的上下文状态。
    """
    client_id: str = Field(..., alias='clientId', description="调用客户端的可验证唯一标识符")
    signature: Optional[str] = Field(None, alias='signature', description="'state'负载的加密签名")
    session_id: Optional[str] = Field(None, alias='sessionId', description="会话或对话标识符")
    state: Dict[str, Any] = Field(..., description="包含上下文数据的自由格式JSON对象")
    
    model_config = ConfigDict(
        populate_by_name=True, 
        extra='forbid'
    )
    
    @property
    def is_verified(self) -> bool:
        """概念上检查护照是否包含有效签名。"""
        return self.signature is not None

2. 消息操作函数

SECURE_PASSPORT_URI = "https://github.com/a2aproject/a2a-samples/tree/main/samples/python/extensions/secure-passport"

def add_secure_passport(message: A2AMessage, context: CallerContext) -> None:
    """将安全护照（CallerContext）添加到消息的元数据中。"""
    message.metadata[SECURE_PASSPORT_URI] = context.model_dump(by_alias=True, exclude_none=True)

def get_secure_passport(message: A2AMessage) -> Optional[CallerContext]:
    """从消息元数据中检索和验证安全护照。"""
    passport_data = message.metadata.get(SECURE_PASSPORT_URI)
    if not passport_data:
        return None

    try:
        return CallerContext.model_validate(deepcopy(passport_data))
    except ValidationError as e:
        import logging
        logging.warning(f"错误：接收到格式错误的安全护照数据。忽略负载：{e}")
        return None

中间件集成模式

1. 客户端中间件

@staticmethod
def client_middleware(next_handler: Callable[[A2AMessage], Any], message: A2AMessage, context: CallerContext):
    """
    [概念中间件层：客户端/调用代理]
    """
    print(f"[中间件：客户端] 为 {context.client_id} 附加安全护照")
    add_secure_passport(message, context)
    return next_handler(message)

2. 服务器中间件

@staticmethod
def server_middleware(next_handler: Callable[[A2AMessage, Optional[CallerContext]], Any], message: A2AMessage):
    """
    [概念中间件层：服务器/接收代理]
    """
    passport = get_secure_passport(message)
    
    if passport:
        print(f"[中间件：服务器] 提取安全护照。已验证：{passport.is_verified}")
    else:
        print("[中间件：服务器] 未找到安全护照或验证失败。")
        
    return next_handler(message, passport)

AgentCard 声明生成

class SecurePassportExtension:
    @staticmethod
    def get_agent_card_declaration(supported_state_keys: Optional[List[str]] = None) -> Dict[str, Any]:
        """
        生成在A2A AgentCard中声明对此扩展支持的JSON结构。
        """
        declaration = {
            "uri": SECURE_PASSPORT_URI,
            "params": {}
        }
        if supported_state_keys:
            declaration["params"]["supportedStateKeys"] = supported_state_keys
        
        return declaration

# 使用示例
# 场景1：代理支持基本安全护照
simple_declaration = SecurePassportExtension.get_agent_card_declaration()

# 场景2：代理支持特定键（如旅行代理）
travel_keys = ["destination", "loyalty_tier", "dates"]
complex_declaration = SecurePassportExtension.get_agent_card_declaration(travel_keys)

完整使用案例演示

1. 货币转换示例

def demonstrate_currency_conversion():
    """演示货币转换使用案例"""
    passport = CallerContext(
        client_id="a2a://travel-orchestrator.com",
        state={"user_preferred_currency": "GBP", "user_id": "U001"},
        signature="sig-currency-1"
    )
    
    message = A2AMessage()
    add_secure_passport(message, passport)
    retrieved = get_secure_passport(message)
    
    if retrieved and retrieved.is_verified:
        currency = retrieved.state.get("user_preferred_currency")
        print(f"以 {currency} 显示价格")

2. 个性化旅行预订示例

def demonstrate_travel_booking():
    """演示个性化旅行预订使用案例"""
    passport = CallerContext(
        client_id="a2a://travel-portal.com",
        session_id="travel-booking-session-999",
        state={
            "destination": "Bali, Indonesia",
            "loyalty_tier": "Platinum"
        },
        signature="sig-travel-2"
    )
    
    message = A2AMessage()
    add_secure_passport(message, passport)
    retrieved = get_secure_passport(message)
    
    if retrieved and retrieved.is_verified:
        tier = retrieved.state.get("loyalty_tier")
        if tier == "Platinum":
            print("为白金会员提供特殊优惠")

运行和测试

1. 运行单元测试

pytest tests/

2. 运行中间件演示

python run.py

5. 最佳实践

安全考虑和签名验证

1. 签名验证策略

def verify_signature(passport: CallerContext, public_key: str) -> bool:
    """
    验证安全护照的数字签名
    """
    if not passport.signature:
        return False
    
    try:
        # 使用公钥验证签名
        # 这里应该实现实际的加密验证逻辑
        return verify_cryptographic_signature(
            data=passport.state,
            signature=passport.signature,
            public_key=public_key
        )
    except Exception as e:
        logging.error(f"签名验证失败：{e}")
        return False

2. 敏感数据处理

• 不要在 state 对象中包含敏感或可变数据，除非实现了强大的端到端加密验证
• 对于高权限操作，应该验证提供的 signature
• 使用适当的加密算法和密钥管理

实现模式和 SDK 助手

1. 中间件集成模式

class SecurePassportMiddleware:
    """安全护照中间件的完整实现"""
    
    def __init__(self, private_key: str = None, public_keys: Dict[str, str] = None):
        self.private_key = private_key
        self.public_keys = public_keys or {}
    
    def client_side(self, message: A2AMessage, context: CallerContext):
        """客户端中间件处理"""
        if self.private_key:
            context.signature = self._sign_context(context.state)
        add_secure_passport(message, context)
    
    def server_side(self, message: A2AMessage) -> Optional[CallerContext]:
        """服务器端中间件处理"""
        passport = get_secure_passport(message)
        if passport and passport.signature:
            client_key = self.public_keys.get(passport.client_id)
            if client_key and not self._verify_signature(passport, client_key):
                logging.warning(f"来自 {passport.client_id} 的签名验证失败")
                return None
        return passport

2. SDK 助手方法

class A2ASDK:
    """A2A SDK 的便利方法"""
    
    @staticmethod
    def create_agent_card_with_passport(supported_keys: List[str] = None) -> Dict:
        """自动生成包含安全护照声明的 AgentCard"""
        return {
            "capabilities": {
                "extensions": [
                    SecurePassportExtension.get_agent_card_declaration(supported_keys)
                ]
            }
        }
    
    @staticmethod
    def send_message_with_passport(message: A2AMessage, context: CallerContext):
        """发送带有安全护照的消息"""
        add_secure_passport(message, context)
        return send_message(message)

错误处理策略

1. 验证错误处理

def safe_get_passport(message: A2AMessage) -> Optional[CallerContext]:
    """安全地获取护照，包含完整的错误处理"""
    try:
        passport = get_secure_passport(message)
        if passport:
            # 记录护照信息（不包含敏感数据）
            logging.info(f"接收到来自 {passport.client_id} 的安全护照")
        return passport
    except ValidationError as e:
        logging.error(f"护照验证失败：{e}")
        return None
    except Exception as e:
        logging.error(f"获取护照时发生意外错误：{e}")
        return None

2. 降级策略

def handle_passport_failure(message: A2AMessage, fallback_handler):
    """当护照处理失败时的降级策略"""
    passport = get_secure_passport(message)
    if not passport:
        logging.warning("未找到有效护照，使用降级处理")
        return fallback_handler(message)
    
    if not passport.is_verified:
        logging.warning("护照未验证，谨慎处理")
        return cautious_handler(message, passport)
    
    return normal_handler(message, passport)

性能优化技巧

1. 缓存策略

class PassportCache:
    """护照缓存，避免重复验证"""
    
    def __init__(self, ttl: int = 300):  # 5分钟TTL
        self.cache = {}
        self.ttl = ttl
    
    def get_verified_passport(self, client_id: str, signature: str) -> Optional[CallerContext]:
        """从缓存获取已验证的护照"""
        cache_key = f"{client_id}:{signature}"
        if cache_key in self.cache:
            cached_data, timestamp = self.cache[cache_key]
            if time.time() - timestamp < self.ttl:
                return cached_data
            else:
                del self.cache[cache_key]
        return None
    
    def cache_verified_passport(self, passport: CallerContext):
        """缓存已验证的护照"""
        cache_key = f"{passport.client_id}:{passport.signature}"
        self.cache[cache_key] = (passport, time.time())

2. 异步处理

import asyncio
from typing import AsyncGenerator

async def process_messages_with_passport(messages: AsyncGenerator[A2AMessage, None]):
    """异步处理带有护照的消息"""
    async for message in messages:
        passport = get_secure_passport(message)
        if passport:
            # 异步验证签名
            is_verified = await verify_signature_async(passport)
            if is_verified:
                await process_verified_message(message, passport)
            else:
                await process_unverified_message(message, passport)
        else:
            await process_no_passport_message(message)

调试和监控方法

1. 详细日志记录

import logging
import json

def setup_passport_logging():
    """设置安全护照的详细日志记录"""
    logger = logging.getLogger('secure_passport')
    logger.setLevel(logging.DEBUG)
    
    handler = logging.StreamHandler()
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    handler.setFormatter(formatter)
    logger.addHandler(handler)

def log_passport_operation(operation: str, passport: CallerContext, success: bool):
    """记录护照操作"""
    logger = logging.getLogger('secure_passport')
    log_data = {
        'operation': operation,
        'client_id': passport.client_id,
        'has_signature': passport.signature is not None,
        'session_id': passport.session_id,
        'state_keys': list(passport.state.keys()),
        'success': success
    }
    logger.info(f"护照操作：{json.dumps(log_data)}")

2. 监控指标

from collections import defaultdict
import time

class PassportMetrics:
    """安全护照的监控指标"""
    
    def __init__(self):
        self.operation_counts = defaultdict(int)
        self.verification_times = []
        self.error_counts = defaultdict(int)
    
    def record_operation(self, operation: str, duration: float, success: bool):
        """记录操作指标"""
        self.operation_counts[operation] += 1
        if operation == 'verify_signature':
            self.verification_times.append(duration)
        
        if not success:
            self.error_counts[operation] += 1
    
    def get_stats(self) -> Dict:
        """获取统计信息"""
        avg_verification_time = (
            sum(self.verification_times) / len(self.verification_times)
            if self.verification_times else 0
        )
        
        return {
            'total_operations': sum(self.operation_counts.values()),
            'operation_breakdown': dict(self.operation_counts),
            'average_verification_time': avg_verification_time,
            'error_breakdown': dict(self.error_counts)
        }

3. 健康检查

def health_check_passport_system() -> Dict[str, Any]:
    """安全护照系统的健康检查"""
    health_status = {
        'status': 'healthy',
        'checks': {}
    }
    
    # 检查护照验证功能
    try:
        test_passport = CallerContext(
            client_id="test://health-check",
            state={"test": "data"},
            signature="test-signature"
        )
        health_status['checks']['passport_creation'] = 'ok'
    except Exception as e:
        health_status['checks']['passport_creation'] = f'error: {e}'
        health_status['status'] = 'unhealthy'
    
    # 检查消息操作功能
    try:
        message = A2AMessage()
        add_secure_passport(message, test_passport)
        retrieved = get_secure_passport(message)
        health_status['checks']['message_operations'] = 'ok' if retrieved else 'error'
    except Exception as e:
        health_status['checks']['message_operations'] = f'error: {e}'
        health_status['status'] = 'unhealthy'
    
    return health_status

总结

安全护照扩展为 A2A 协议提供了一个强大而灵活的上下文共享机制。通过遵循本指南中的最佳实践，您可以：

1. 实现即时个性化：让代理能够立即了解调用者的上下文
2. 确保安全通信：通过数字签名验证数据的完整性和来源
3. 优化性能：减少不必要的多轮对话和验证开销
4. 支持多种场景：从低风险的零售协助到高安全性的数据库访问

这个扩展为构建更加智能、高效和安全的代理生态系统奠定了坚实的基础。

A2A 扩展列表

• A2A 可追溯性扩展
• A2A 时间戳扩展
• A2A 安全护照扩展