python-a2a

The examples/streaming/ directory provides a progression of implementations for different streaming use cases:

• basic_streaming.py: Minimal implementation of a streaming server and client with proper dictionary chunk handling.
• 01_basic_streaming.py: Comprehensive introduction covering simulated thinking delays and natural language chunking.
• 02_advanced_streaming.py: Focuses on metrics tracking, performance visualization, and various chunking strategies (sentence, word, paragraph).
• 03_streaming_llm_integration.py: Demonstrates bridging LLM provider APIs (OpenAI, Anthropic, Bedrock) with A2A using transformer pipelines.
• 04_task_based_streaming.py: Implements structured task state transitions (e.g., WAITING, COMPLETED) and artifact generation.
• 05_streaming_ui_integration.py: Shows CLI-based visualization and Web interfaces using Server-Sent Events (SSE) with interactive controls (pause, resume, cancel).
• 06_distributed_streaming.py: Demonstrates distributed architecture with multiple servers, load balancing, and stream aggregation.

The examples/ directory is organized into several functional categories to help you find specific implementation patterns:

• Getting Started: Beginner-level examples for creating messages (hello_a2a.py), connecting as a client (simple_client.py), building a server (simple_server.py), and implementing function calling (function_calling.py).
• Streaming: Real-time response implementations, ranging from basic streaming to advanced techniques like metrics tracking, LLM integration, task-based progress, and UI/distributed streaming.
• Building Blocks: Core protocol concepts including Agent Discovery (agent cards), Messages and Conversations, the Task model, and defining Agent Skills using decorators.
• AI-Powered Agents: Integration examples for OpenAI, Anthropic, and AWS Bedrock.
• Agent Network: Orchestration patterns including basic/parallel workflows, smart routing, and automatic agent discovery.
• Model Context Protocol (MCP): Extending agents with external tools via GitHub, Browserbase, or Filesystem MCP providers.
• Complete Applications: End-to-end use cases like a Weather Assistant or an OpenAI Travel Planner.
• Developer Tools: CLI tools, interactive documentation generation, and agent testing utilities.

Python A2A provides a discovery mechanism to allow agents to find each other in a network.

1. AgentRegistry: A central server that maintains a registry of available agents.
2. AgentCard: A metadata object describing an agent's name, description, URL, version, and capabilities.
3. enable_discovery: A function used by an A2AServer to register itself with a registry.
4. DiscoveryClient: A client used to query a registry and discover available agents.

from python_a2a import AgentCard, A2AServer, run_server
from python_a2a.discovery import AgentRegistry, run_registry, enable_discovery, DiscoveryClient

# 1. Setup Registry
registry = AgentRegistry(name="A2A Registry Server")
# (Run registry in a thread using run_registry...)

# 2. Create and Register an Agent
agent_card = AgentCard(
    name="Weather Agent",
    url="http://localhost:8001",
    capabilities={"weather_forecasting": True}
)
agent = A2AServer(agent_card=agent_card)
enable_discovery(agent, registry_url="http://localhost:8000")

# 3. Discover Agents via Client
client = DiscoveryClient()
client.add_registry("http://localhost:8000")
agents = client.discover()
for agent in agents:
    print(f"Found {agent.name} at {agent.url}")

You can build complex systems by creating specialized agents that communicate with each other using A2AClient. In a multi-agent architecture:

1. Specialized Agents: Define agents using the @agent decorator and inherit from A2AServer. Use the @skill decorator to expose specific capabilities (e.g., get_weather).
2. Inter-Agent Communication: One agent can act as a client to another by initializing A2AClient("http://<other-agent-url>"). Use client.ask(query) to send requests to the specialized agent.
3. Orchestration: Create a 'UI Agent' or 'Orchestrator Agent' that receives user input and routes requests to the appropriate specialized agents based on the query content.

To run a multi-agent system, start each agent's server on a different port using run_server(agent, port=XXXX).

# Example of an Orchestrator Agent routing to a specialized agent
from python_a2a import A2AServer, A2AClient, run_server

class AssistantAgent(A2AServer):
    def __init__(self):
        super().__init__()
        self.weather_client = A2AClient("http://localhost:5001")

    def handle_task(self, task):
        text = task.message.get("content", {}).get("text", "")
        if "weather" in text.lower():
            # Route to the weather agent
            response = self.weather_client.ask(text)
            task.artifacts = [{"parts": [{"type": "text", "text": response}]}]
        # ... set task status to COMPLETED
        return task

if __name__ == "__main__":
    run_server(AssistantAgent(), port=5000)

You can connect to multiple MCP servers simultaneously by providing a configuration dictionary to a FastMCPAgent. This allows an agent to orchestrate tools from different sources (e.g., a local calculator, a remote database, and a filesystem server) in a single task.

# Connect to multiple MCP servers simultaneously
mcp_config = {
    "calculator": {"command": ["python", "calc_server.py"]},
    "database": {"url": "https://db.example.com/mcp/sse"},
    "files": {"command": ["node", "file_server.js"]}
}

class MultiToolAgent(A2AServer, FastMCPAgent):
    def __init__(self):
        FastMCPAgent.__init__(self, mcp_servers=mcp_config)
    
    async def solve_complex_task(self, query):
        # Use tools from different servers
        calc_result = await self.call_mcp_tool("calculator", "add", a=5, b=3)
        data = await self.call_mcp_tool("database", "query", sql="SELECT * FROM users")
        file_content = await self.call_mcp_tool("files", "read", path="/config.json")
        
        return self._combine_results(calc_result, data, file_content)

Python A2A provides a discovery mechanism for agent ecosystems.

1. AgentRegistry: A central server that stores AgentCard information.
2. AgentCard: A metadata object containing an agent's name, description, URL, version, and capabilities.
3. enable_discovery: A function used on an A2AServer to register it with a registry.
4. DiscoveryClient: A client used to connect to a registry and discover available agents.

from python_a2a import AgentCard, A2AServer, run_server
from python_a2a.discovery import AgentRegistry, run_registry, enable_discovery, DiscoveryClient

# 1. Setup Registry
registry = AgentRegistry(name="A2A Registry Server")
# run_registry(registry, host="0.0.0.0", port=8000)

# 2. Create Agent with Card
agent_card = AgentCard(
    name="Weather Agent",
    url="http://localhost:8001",
    capabilities={"weather_forecasting": True}
)
agent = A2AServer(agent_card=agent_card)

# 3. Enable Discovery (registers agent to registry)
discovery_client = enable_discovery(agent, registry_url="http://localhost:8000")

# 4. Use DiscoveryClient to find agents
client = DiscoveryClient()
client.add_registry("http://localhost:8000")
agents = client.discover()
for agent in agents:
    print(f"{agent.name} at {agent.url}")