Python A2A

The Definitive Python Implementation of Google's Agent-to-Agent (A2A) Protocol with Model Context Protocol (MCP) Integration

🌟 Overview

Python A2A is a comprehensive, production-ready library for implementing Google's Agent-to-Agent (A2A) protocol with full support for the Model Context Protocol (MCP). It provides everything you need to build interoperable AI agent ecosystems that can collaborate seamlessly to solve complex problems.

The A2A protocol establishes a standard communication format that enables AI agents to interact regardless of their underlying implementation, while MCP extends this capability by providing a standardized way for agents to access external tools and data sources. Python A2A makes these protocols accessible with an intuitive API that developers of all skill levels can use to build sophisticated multi-agent systems.

📋 What's New in v0.5.X

• 🔌 MCP v2.0 Complete Rewrite: Rebuilt MCP implementation from scratch following JSON-RPC 2.0 specification
• 🏗️ Provider Architecture: New provider-based architecture for external MCP servers with GitHub, Browserbase, and Filesystem providers
• 🚀 Real-World MCP Examples: Production-ready examples with actual services (no mocks!) including GitHub, browser automation, and file management
• 🛡️ Enterprise MCP Support: Robust transport abstraction supporting stdio and SSE for production deployments
• 🔄 Backward Compatible Migration: Seamless upgrade path from previous MCP implementations
• Agent Flow UI: Visual workflow editor for building and managing agent networks with drag-and-drop interface
• Agent Discovery: Built-in support for agent registry and discovery with full Google A2A protocol compatibility
• LangChain Integration: Seamless integration with LangChain's tools and agents
• Expanded Tool Ecosystem: Use tools from both LangChain and MCP in any agent
• Enhanced Agent Interoperability: Convert between A2A agents and LangChain agents
• Mixed Workflow Engine: Build workflows combining both ecosystems
• Simplified Agent Development: Access thousands of pre-built tools instantly
• Advanced Streaming Architecture: Enhanced streaming with Server-Sent Events (SSE), better error handling, and robust fallback mechanisms
• Task-Based Streaming: New tasks_send_subscribe method for streaming task updates in real-time
• Streaming Chunks API: Improved chunk processing with the StreamingChunk class for structured streaming data
• Multi-Endpoint Support: Automatic discovery and fallback across multiple streaming endpoints

📋 What's New in v0.4.X

• Agent Network System: Manage and discover multiple agents with the new AgentNetwork class
• Real-time Streaming: Implement streaming responses with StreamingClient for responsive UIs
• Workflow Engine: Define complex multi-agent workflows using the new fluent API with conditional branching and parallel execution
• AI-Powered Router: Automatically route queries to the most appropriate agent with the AIAgentRouter
• Command Line Interface: Control your agents from the terminal with the new CLI tool
• Enhanced Asynchronous Support: Better async/await support throughout the library
• New Connection Options: Improved error handling and retry logic for more robust agent communication

✨ Why Choose Python A2A?

• Complete Implementation: Fully implements the official A2A specification with zero compromises
• Agent Discovery: Built-in agent registry and discovery for building agent ecosystems
• MCP Integration: First-class support for Model Context Protocol for powerful tool-using agents
• Enterprise Ready: Built for production environments with robust error handling and validation
• Framework Agnostic: Works with any Python framework (Flask, FastAPI, Django, etc.)
• LLM Provider Flexibility: Native integrations with OpenAI, Anthropic, AWS Bedrock, Ollama, and more
• Minimal Dependencies: Core functionality requires only the requests library
• Excellent Developer Experience: Comprehensive documentation, type hints, and examples

📦 Installation

Using pip (traditional)

Install the base package with all dependencies:

pip install python-a2a  # Includes LangChain, MCP, and other integrations

Or install with specific components based on your needs:

# For Flask-based server support
pip install "python-a2a[server]"

# For OpenAI integration
pip install "python-a2a[openai]"

# For Anthropic Claude integration
pip install "python-a2a[anthropic]"

# For AWS-Bedrock integration
pip install "python-a2a[bedrock]"

# For Ollama integration  
pip install "python-a2a[ollama]"

# For MCP support (Model Context Protocol)
pip install "python-a2a[mcp]"

# For all optional dependencies
pip install "python-a2a[all]"

Using UV (recommended)

UV is a modern Python package management tool that's faster and more reliable than pip. To install with UV:

# Install UV if you don't have it already
curl -LsSf https://astral.sh/uv/install.sh | sh

# Install the base package
uv install python-a2a

Development Installation

For development, UV is recommended for its speed:

# Clone the repository
git clone https://github.com/themanojdesai/python-a2a.git
cd python-a2a

# Create a virtual environment and install development dependencies
uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
uv pip install -e ".[dev]"

💡 Tip: Click the code blocks to copy them to your clipboard.

🚀 Quick Start Examples

1. Create a Simple A2A Agent with Skills

from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState

@agent(
    name="Weather Agent",
    description="Provides weather information",
    version="1.0.0"
)
class WeatherAgent(A2AServer):
    
    @skill(
        name="Get Weather",
        description="Get current weather for a location",
        tags=["weather", "forecast"]
    )
    def get_weather(self, location):
        """Get weather for a location."""
        # Mock implementation
        return f"It's sunny and 75°F in {location}"
    
    def handle_task(self, task):
        # Extract location from message
        message_data = task.message or {}
        content = message_data.get("content", {})
        text = content.get("text", "") if isinstance(content, dict) else ""
        
        if "weather" in text.lower() and "in" in text.lower():
            location = text.split("in", 1)[1].strip().rstrip("?.")
            
            # Get weather and create response
            weather_text = self.get_weather(location)
            task.artifacts = [{
                "parts": [{"type": "text", "text": weather_text}]
            }]
            task.status = TaskStatus(state=TaskState.COMPLETED)
        else:
            task.status = TaskStatus(
                state=TaskState.INPUT_REQUIRED,
                message={"role": "agent", "content": {"type": "text", 
                         "text": "Please ask about weather in a specific location."}}
            )
        return task

# Run the server
if __name__ == "__main__":
    agent = WeatherAgent()
    run_server(agent, port=5000)

2. Build an Agent Network with Multiple Agents

from python_a2a import AgentNetwork, A2AClient, AIAgentRouter

# Create an agent network
network = AgentNetwork(name="Travel Assistant Network")

# Add agents to the network
network.add("weather", "http://localhost:5001")
network.add("hotels", "http://localhost:5002")
network.add("attractions", "http://localhost:5003")

# Create a router to intelligently direct queries to the best agent
router = AIAgentRouter(
    llm_client=A2AClient("http://localhost:5000/openai"),  # LLM for making routing decisions
    agent_network=network
)

# Route a query to the appropriate agent
agent_name, confidence = router.route_query("What's the weather like in Paris?")
print(f"Routing to {agent_name} with {confidence:.2f} confidence")

# Get the selected agent and ask the question
agent = network.get_agent(agent_name)
response = agent.ask("What's the weather like in Paris?")
print(f"Response: {response}")

# List all available agents
print("\nAvailable Agents:")
for agent_info in network.list_agents():
    print(f"- {agent_info['name']}: {agent_info['description']}")

Real-time Streaming

Get real-time responses from agents with comprehensive streaming support:

import asyncio
from python_a2a import StreamingClient, Message, TextContent, MessageRole

async def main():
    client = StreamingClient("http://localhost:5000")
    
    # Create a message with required role parameter
    message = Message(
        content=TextContent(text="Tell me about A2A streaming"),
        role=MessageRole.USER
    )
    
    # Stream the response and process chunks in real-time
    try:
        async for chunk in client.stream_response(message):
            # Handle different chunk formats (string or dictionary)
            if isinstance(chunk, dict):
                if "content" in chunk:
                    print(chunk["content"], end="", flush=True)
                elif "text" in chunk:
                    print(chunk["text"], end="", flush=True)
                else:
                    print(str(chunk), end="", flush=True)
            else:
                print(str(chunk), end="", flush=True)
    except Exception as e:
        print(f"Streaming error: {e}")

Check out the examples/streaming/ directory for complete streaming examples:

• basic_streaming.py: Minimal streaming implementation (start here!)
• 01_basic_streaming.py: Comprehensive introduction to streaming basics
• 02_advanced_streaming.py: Advanced streaming with different chunking strategies
• 03_streaming_llm_integration.py: Integrating streaming with LLM providers
• 04_task_based_streaming.py: Task-based streaming with progress tracking
• 05_streaming_ui_integration.py: Streaming UI integration (CLI and web)
• 06_distributed_streaming.py: Distributed streaming architecture

3. Workflow Engine

The new workflow engine allows you to define complex agent interactions:

from python_a2a import AgentNetwork, Flow
import asyncio

async def main():
    # Set up agent network
    network = AgentNetwork()
    network.add("research", "http://localhost:5001")
    network.add("summarize