a2a-protocols

What this skill does

# Agent-to-Agent (A2A) Protocols

**Enable agents to discover, communicate, and collaborate across frameworks and systems.**

## The Interoperability Challenge

Different agent frameworks:
- LangChain/LangGraph
- AutoGen
- CrewAI
- Custom implementations

**Problem**: They can't talk to each other natively.

**Solution**: Standard protocols for agent communication.

## Protocol Landscape (2026)

| Protocol | Purpose | Adoption |
|----------|---------|----------|
| **MCP** (Model Context Protocol) | Tool/resource sharing | High (Anthropic-backed) |
| **A2A** (Agent-to-Agent) | Agent coordination | Growing |
| **OpenAI Agents Protocol** | Agent invocation | OpenAI ecosystem |
| **Custom REST/gRPC** | Point-to-point | Common |

## MCP Integration

### MCP Overview

MCP standardizes how agents access tools and resources:

```
┌─────────────────┐          ┌─────────────────┐
│     Agent       │          │   MCP Server    │
│  (MCP Client)   │◀────────▶│  (Tools/Data)   │
└─────────────────┘          └─────────────────┘
         │
         │  MCP Protocol
         │  - List tools
         │  - Call tools
         │  - Access resources
         ▼
┌─────────────────┐
│  Another Agent  │
│  (MCP Client)   │
└─────────────────┘
```

### Creating an MCP Server

```python
from mcp.server import Server
from mcp.types import Tool, TextContent

# Create MCP server
server = Server("my-agent-tools")

@server.tool()
async def search_database(query: str) -> str:
    """Search the internal database."""
    results = await db.search(query)
    return json.dumps(results)

@server.tool()
async def send_notification(
    recipient: str,
    message: str
) -> str:
    """Send a notification to a user."""
    await notifications.send(recipient, message)
    return "Notification sent"

# Run server
if __name__ == "__main__":
    server.run()
```

### Connecting Agent to MCP

```python
from mcp import ClientSession, StdioServerParameters
from langchain_mcp import MCPToolkit

async def create_mcp_agent():
    """Create agent with MCP tools."""

    # Connect to MCP server
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_server.py"]
    )

    async with ClientSession(server_params) as session:
        # Get tools from MCP
        toolkit = MCPToolkit(session=session)
        tools = toolkit.get_tools()

        # Create LangChain agent with MCP tools
        agent = create_react_agent(llm, tools)

        return agent
```

## Agent-to-Agent Communication

### Message-Based Protocol

```python
from dataclasses import dataclass
from enum import Enum
from typing import Any
import uuid

class MessageType(Enum):
    REQUEST = "request"
    RESPONSE = "response"
    BROADCAST = "broadcast"
    HANDOFF = "handoff"

@dataclass
class AgentMessage:
    """Standard message format for A2A communication."""
    id: str
    type: MessageType
    sender: str
    recipient: str | None  # None for broadcasts
    content: dict
    reply_to: str | None = None
    timestamp: str = None

    def __post_init__(self):
        if not self.id:
            self.id = str(uuid.uuid4())
        if not self.timestamp:
            self.timestamp = datetime.now().isoformat()

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "type": self.type.value,
            "sender": self.sender,
            "recipient": self.recipient,
            "content": self.content,
            "reply_to": self.reply_to,
            "timestamp": self.timestamp
        }

    @classmethod
    def from_dict(cls, data: dict) -> "AgentMessage":
        return cls(
            id=data["id"],
            type=MessageType(data["type"]),
            sender=data["sender"],
            recipient=data.get("recipient"),
            content=data["content"],
            reply_to=data.get("reply_to"),
            timestamp=data.get("timestamp")
        )
```

### Agent Registry

```python
class AgentRegistry:
    """Central registry for agent discovery."""

    def __init__(self, redis_client):
        self.redis = redis_client

    async def register(
        self,
        agent_id: str,
        capabilities: list[str],
        endpoint: str,
        metadata: dict = None
    ):
        """Register an agent with its capabilities."""
        agent_info = {
            "agent_id": agent_id,
            "capabilities": capabilities,
            "endpoint": endpoint,
            "metadata": metadata or {},
            "registered_at": datetime.now().isoformat(),
            "last_heartbeat": datetime.now().isoformat()
        }

        # Store agent info
        await self.redis.hset(
            "agents",
            agent_id,
            json.dumps(agent_info)
        )

        # Index by capability
        for cap in capabilities:
            await self.redis.sadd(f"capability:{cap}", agent_id)

    async def find_by_capability(self, capability: str) -> list[dict]:
        """Find agents with a specific capability."""
        agent_ids = await self.redis.smembers(f"capability:{capability}")
        agents = []
        for aid in agent_ids:
            info = await self.redis.hget("agents", aid)
            if info:
                agents.append(json.loads(info))
        return agents

    async def heartbeat(self, agent_id: str):
        """Update agent heartbeat."""
        info = await self.redis.hget("agents", agent_id)
        if info:
            data = json.loads(info)
            data["last_heartbeat"] = datetime.now().isoformat()
            await self.redis.hset("agents", agent_id, json.dumps(data))
```

### Inter-Agent Communication

```python
class AgentCommunicator:
    """Handle agent-to-agent communication."""

    def __init__(self, agent_id: str, registry: AgentRegistry):
        self.agent_id = agent_id
        self.registry = registry
        self.message_handlers = {}

    def on_message(self, message_type: MessageType):
        """Decorator to register message handlers."""
        def decorator(func):
            self.message_handlers[message_type] = func
            return func
        return decorator

    async def send(
        self,
        recipient: str,
        content: dict,
        message_type: MessageType = MessageType.REQUEST
    ) -> AgentMessage:
        """Send message to another agent."""
        # Get recipient endpoint
        recipient_info = await self.registry.get(recipient)
        if not recipient_info:
            raise ValueError(f"Agent {recipient} not found")

        message = AgentMessage(
            id=str(uuid.uuid4()),
            type=message_type,
            sender=self.agent_id,
            recipient=recipient,
            content=content
        )

        # Send via HTTP
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{recipient_info['endpoint']}/messages",
                json=message.to_dict()
            ) as response:
                return await response.json()

    async def request(
        self,
        capability: str,
        content: dict,
        timeout: float = 30.0
    ) -> dict:
        """Find agent by capability and send request."""
        agents = await self.registry.find_by_capability(capability)
        if not agents:
            raise ValueError(f"No agent with capability: {capability}")

        # Simple: pick first available
        target = agents[0]

        response = await self.send(
            target["agent_id"],
            content,
            MessageType.REQUEST
        )

        return response

    async def broadcast(self, content: dict, capability: str = None):
        """Broadcast message to multiple agents."""
        if capability:
            agents = await self.registry.find_by_capability(capability)
        else:
            agents = await self.registry.get_all()

        tasks = [
            self.send(a["agent_id"], content, MessageType.BROADCAST)
            for a in agents
        ]

        return a