1  """
 2  Utility functions for A2A protocol operations.
 3  """
 4  
 5  from typing import Dict, Any, Optional
 6  import logging
 7  
 8  from a2a.types import AgentCard
 9  
10  log = logging.getLogger(__name__)
11  
12  
13  def is_gateway_card(agent_card: AgentCard) -> bool:
14      """
15      Check if an AgentCard represents a gateway.
16  
17      Gateways are identified by the presence of the gateway-role extension:
18      https://solace.com/a2a/extensions/sam/gateway-role
19  
20      Args:
21          agent_card: The AgentCard to check
22  
23      Returns:
24          True if this is a gateway card, False otherwise (agent card)
25      """
26      if not agent_card:
27          return False
28  
29      if not agent_card.capabilities:
30          return False
31  
32      # Handle both dict and AgentCapabilities object
33      if isinstance(agent_card.capabilities, dict):
34          extensions = agent_card.capabilities.get("extensions")
35      else:
36          extensions = agent_card.capabilities.extensions
37  
38      if not extensions:
39          return False
40  
41      for ext in extensions:
42          ext_uri = ext.uri if hasattr(ext, 'uri') else ext.get('uri')
43          if ext_uri == "https://solace.com/a2a/extensions/sam/gateway-role":
44              return True
45  
46      return False
47  
48  
49  def extract_gateway_info(agent_card: AgentCard) -> Optional[Dict[str, Any]]:
50      """
51      Extract gateway-specific information from AgentCard extensions.
52  
53      Extracts information from:
54      - gateway-role extension: gateway_id, gateway_type, namespace
55      - deployment extension: deployment_id (optional)
56  
57      Args:
58          agent_card: The AgentCard to extract info from
59  
60      Returns:
61          Dict with gateway_id, gateway_type, namespace, and optionally deployment_id
62          Returns None if not a gateway card
63      """
64      if not is_gateway_card(agent_card):
65          return None
66  
67      info = {}
68  
69      # Handle both dict and AgentCapabilities object
70      if isinstance(agent_card.capabilities, dict):
71          extensions = agent_card.capabilities.get("extensions", [])
72      else:
73          extensions = agent_card.capabilities.extensions or []
74  
75      for ext in extensions:
76          # Handle both dict and AgentExtension object
77          ext_uri = ext.uri if hasattr(ext, 'uri') else ext.get('uri')
78          ext_params = ext.params if hasattr(ext, 'params') else ext.get('params', {})
79  
80          if ext_uri == "https://solace.com/a2a/extensions/sam/gateway-role":
81              info.update({
82                  "gateway_id": ext_params.get("gateway_id"),
83                  "gateway_type": ext_params.get("gateway_type"),
84                  "namespace": ext_params.get("namespace"),
85              })
86          elif ext_uri == "https://solace.com/a2a/extensions/sam/deployment":
87              info["deployment_id"] = ext_params.get("deployment_id")
88  
89      return info