Official a2a switch

[destroyersrt]

Official A2A Protocol Support

This PR introduces official Google's A2A protocol support to NANDA, implementing a standardized protocol architecture that enables multi-protocol communication and prepares the framework for future protocol additions.

🎯 Key Changes

Protocol Architecture

• Implemented ProtocolRouter for managing multiple communication protocols
• Added official A2A protocol adapter (protocols/a2a/protocol.py) with standard message format support
• Created extensible BaseProtocol interface for future protocol implementations (SLIM, gRPC, etc.)
• Introduced AgentExecutor for protocol-agnostic agent logic execution

Enhanced Core Components

• Refactored NANDA class to support protocol configuration via protocols parameter
• Updated AgentBridge to work with protocol router instead of direct A2A coupling
• Enhanced RegistryClient with async support and improved registration handling
• Added structured agent metadata (name, domain, specialization, description, capabilities)

Registration & Discovery

• Agents now register with supported_protocols and endpoints metadata
• Registry receives complete agent facts including protocol capabilities

Example Agent Improvements

• Refactored examples/nanda_agent.py to demonstrate new protocol system

📦 Architecture

NANDA (adapter.py)
├── ProtocolRouter (protocols/router.py)
│ ├── A2AProtocol (protocols/a2a/protocol.py)
│ └── [Future: SLIMProtocol, CustomProtocol]
├── AgentBridge (agent_bridge.py)
├── RegistryClient (registry_client.py)
└── AgentExecutor (protocols/AgentExecutor.py)

✅ Benefits

• Standardization: Official A2A protocol compliance for interoperability
• Extensibility: Easy to add new protocols without modifying core
• Future-Ready: Foundation for SLIM protocol and custom protocol support
• Better Metadata: Rich agent information for discovery and routing

🚀 Next Steps

• SLIM protocol implementation
• Custom protocol support
• Protocol negotiation for multi-protocol agents

[ryanRfox]

Here are a few suggested changes for this PR.

[ryanRfox]

remove from PR. retain copies locally.

[ryanRfox]

remove from PR. retain copies locally.

[ryanRfox]

suggest to log a WARNING if verify=FALSE to raise the potential security vulnerability (no SSL cert verification) and mitigate the impacts if used in production.

[ryanRfox]

should await here, like in L61

[ryanRfox]

should await here, like in L61

[ryanRfox]

rename file to agent_executor.py to match python naming conventions.

this will require updates to imports and init

# Change from:
from ..AgentExecutor import NANDAAgentExecutor

# Change to:
from ..agent_executor import NANDAAgentExecutor

[ryanRfox]

Missing Resource Cleanup

Existing Code

a2a/protocol.py:78:

class A2AProtocol(AgentProtocol):
    def __init__(self, ...):
        # ... init ...

        # HTTP client for A2A client
        self.httpx_client = httpx.AsyncClient()  # ❌ Never closed

adapter.py:170-174:

def stop(self):
    """Stop the agent and cleanup telemetry"""
    if self.telemetry:
        self.telemetry.stop()  # ✅ Cleans up telemetry
    print(f"🛑 Stopping agent '{self.agent_id}'")
    # ❌ But doesn't close HTTP clients or other resources

[ryanRfox]

client never closed. Suggest to make A2AProtocol a proper async context manager.

import httpx
from typing import Optional


class A2AProtocol(AgentProtocol):
    """A2A SDK protocol adapter"""

    def __init__(self, ...):
        """Initialize A2A protocol adapter"""
        self.agent_id = agent_id
        self.agent_name = agent_name
        self.public_url = public_url
        # ... other init ...

        # HTTP client (will be initialized on first use or in __aenter__)
        self.httpx_client: Optional[httpx.AsyncClient] = None
        self._client_managed_externally = False

    async def _ensure_client(self) -> httpx.AsyncClient:
        """Ensure HTTP client is initialized"""
        if self.httpx_client is None:
            self.httpx_client = httpx.AsyncClient()
        return self.httpx_client

    async def send_message(self, target_url: str, message: Dict[str, Any]) -> Dict[str, Any]:
        """Send message using A2A client"""
        client = await self._ensure_client()

        try:
            # ... existing send logic using client ...
            pass
        except Exception as e:
            logger.exception(f"Error sending message: {e}")
            raise

    async def close(self) -> None:
        """Close HTTP client and cleanup resources"""
        if self.httpx_client is not None and not self._client_managed_externally:
            await self.httpx_client.aclose()
            self.httpx_client = None
            logger.debug(f"Closed HTTP client for {self.agent_id}")

    async def __aenter__(self):
        """Async context manager entry"""
        self.httpx_client = httpx.AsyncClient()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit"""
        await self.close()

    def get_protocol_name(self) -> str:
        """Return protocol identifier"""
        return "a2a"

Update ProtocolRouter to manage protocol lifecycle:

class ProtocolRouter:
    """Manages multiple protocol adapters and routes messages"""

    def __init__(self) -> None:
        self.protocols: Dict[str, AgentProtocol] = {}
        self.default_protocol: Optional[str] = None

    # ... existing methods ...

    async def close_all(self) -> None:
        """Close all protocol adapters and cleanup resources"""
        logger.info("Closing all protocol adapters...")

        for name, protocol in self.protocols.items():
            try:
                if hasattr(protocol, 'close'):
                    await protocol.close()
                    logger.debug(f"Closed {name} protocol")
            except Exception as e:
                logger.error(f"Error closing {name} protocol: {e}")

    async def __aenter__(self):
        """Async context manager entry"""
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit"""
        await self.close_all()

Update NANDA adapter to properly cleanup:

class NANDA:
    """Simple NANDA class for clean agent deployment"""

    # ... existing __init__ ...

    async def stop(self) -> None:
        """Stop the agent and cleanup all resources"""
        logger.info(f"Stopping agent '{self.agent_id}'...")

        # Close registry client
        if self.registry and hasattr(self.registry, 'close'):
            try:
                await self.registry.close()
                logger.debug("Registry client closed")
            except Exception as e:
                logger.error(f"Error closing registry: {e}")

        # Close all protocol adapters
        if self.router:
            try:
                await self.router.close_all()
                logger.debug("All protocols closed")
            except Exception as e:
                logger.error(f"Error closing protocols: {e}")

        # Stop telemetry
        if self.telemetry:
            try:
                self.telemetry.stop()
                logger.debug("Telemetry stopped")
            except Exception as e:
                logger.error(f"Error stopping telemetry: {e}")

        logger.info(f"Agent '{self.agent_id}' stopped successfully")

    async def __aenter__(self):
        """Async context manager entry"""
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit"""
        await self.stop()