vac_a2a_agent.py

Source: src/sunholo/a2a/vac_a2a_agent.py

Classes

VACA2AAgent

A2A Agent implementation for Sunholo VACs.

This class wraps VAC functionality to be compatible with the Agent-to-Agent protocol, allowing VACs to participate in A2A ecosystems.

• init(self, base_url: str, stream_interpreter: Callable, vac_interpreter: Optional[Callable] = None, vac_names: Optional[List[str]] = None, agent_config: Optional[Dict[str, Any]] = None)
  • Initialize the A2A agent.

Args: base_url: Base URL where the agent is hosted stream_interpreter: Function for streaming VAC interactions vac_interpreter: Function for static VAC interactions (optional) vac_names: List of VAC names to expose (discovers all if None) agent_config: Additional agent configuration

• _create_error_response(self, request_id: Any, error_code: int, error_message: str) -> Dict[str, Any]
  • Create a JSON-RPC error response.

Args: request_id: The request ID error_code: JSON-RPC error code error_message: Error message

Returns: JSON-RPC error response

• _estimate_task_duration(self, skill_name: str, input_data: Dict[str, Any]) -> float
  • Estimate task duration in seconds.

Args: skill_name: Name of the skill input_data: Input parameters

Returns: Estimated duration in seconds

• get_agent_card(self) -> Dict[str, Any]
  • Get the agent card for A2A discovery.

Returns: Agent card dictionary

• get_discovery_endpoints(self) -> Dict[str, str]
  • Get the A2A discovery endpoint paths.

Returns: Dictionary mapping endpoint names to their paths

• get_stats(self) -> Dict[str, Any]
  • Get agent statistics.

Returns: Dictionary with agent statistics

• handle_push_notification_set(self, request_data: Dict[str, Any]) -> Dict[str, Any]
  • Handle A2A push notification settings.

Args: request_data: JSON-RPC request data

Returns: JSON-RPC response data

• handle_task_cancel(self, request_data: Dict[str, Any]) -> Dict[str, Any]
  • Handle A2A task cancel request.

Args: request_data: JSON-RPC request data

Returns: JSON-RPC response data

• handle_task_get(self, request_data: Dict[str, Any]) -> Dict[str, Any]
  • Handle A2A task get request.

Args: request_data: JSON-RPC request data

Returns: JSON-RPC response data

• handle_task_send(self, request_data: Dict[str, Any]) -> Dict[str, Any]
  • Handle A2A task send request.

Args: request_data: JSON-RPC request data

Returns: JSON-RPC response data

• handle_task_send_subscribe(self, request_data: Dict[str, Any]) -> AsyncGenerator[str, NoneType]
  • Handle A2A task send with subscription (SSE).

Args: request_data: JSON-RPC request data

Yields: Server-sent event data strings