Quickstart

Quickstart

This quickstart guide will help you create your first A2A agent and client in just a few minutes.

Installation

First, install the Python A2A library:

pip install python-a2a

Creating Your First A2A Agent

Let's create a simple agent that responds to messages:

from python_a2a import A2AServer, agent, skill, run_server

@agent(
    name="Echo Agent",
    description="Simple agent that echoes messages"
)
class EchoAgent(A2AServer):
    
    @skill(
        name="Echo Message",
        description="Echo back the input message"
    )
    def echo(self, message):
        return f"You said: {message}"
    
    def handle_task(self, task):
        """Process an A2A task by echoing the input."""
        return {"output": self.echo(task.input)}

# Create and run the agent
if __name__ == "__main__":
    echo_agent = EchoAgent()
    run_server(echo_agent, host="localhost", port=5000)

Save this file as echo_agent.py and run it:

python echo_agent.py

Your agent is now running at http://localhost:5000.

Connecting to Your Agent

Now, let's create a client to communicate with your agent:

from python_a2a import HTTPClient

# Create a client that connects to your agent
client = HTTPClient("http://localhost:5000")

# Send a message and get the response
response = client.send_message("Hello, Agent!")
print(response.content)

Save this file as echo_client.py and run it:

python echo_client.py

You should see:

You said: Hello, Agent!

Creating a More Useful Agent

Let's create a more practical agent that provides weather information:

from python_a2a import A2AServer, agent, skill, run_server
import json

@agent(
    name="Weather Agent",
    description="Provides weather information"
)
class WeatherAgent(A2AServer):
    
    # Sample weather data
    weather_data = {
        "new york": {"condition": "Sunny", "temperature": 75},
        "london": {"condition": "Rainy", "temperature": 62},
        "tokyo": {"condition": "Cloudy", "temperature": 80},
        "paris": {"condition": "Partly Cloudy", "temperature": 70},
    }
    
    @skill(
        name="Get Weather",
        description="Get current weather for a location",
        parameters={
            "location": {
                "type": "string",
                "description": "The city to get weather for"
            }
        }
    )
    def get_weather(self, location):
        location = location.lower()
        if location in self.weather_data:
            weather = self.weather_data[location]
            return f"The weather in {location.title()} is {weather['condition']} with a temperature of {weather['temperature']}°F."
        else:
            return f"Sorry, I don't have weather data for {location}."
    
    def handle_task(self, task):
        """Process an A2A task."""
        try:
            # Try to parse the input as JSON
            if isinstance(task.input, str) and task.input.strip().startswith("{"):
                data = json.loads(task.input)
                if "location" in data:
                    return {"output": self.get_weather(data["location"])}
            
            # If not JSON or doesn't have location, treat as direct location query
            return {"output": self.get_weather(task.input)}
        except Exception as e:
            return {"output": f"Error: {str(e)}"}

# Create and run the agent
if __name__ == "__main__":
    weather_agent = WeatherAgent()
    run_server(weather_agent, host="localhost", port=5000)

Save this as weather_agent.py and run it:

python weather_agent.py

Using the Weather Agent

Now let's connect to our weather agent:

from python_a2a import HTTPClient

client = HTTPClient("http://localhost:5000")

# Send a direct query
response = client.send_message("New York")
print(response.content)

# Send a structured query
structured_query = {"location": "Tokyo"}
response = client.send_message(structured_query)
print(response.content)

Save this as weather_client.py and run it:

python weather_client.py

You should see the weather information for New York and Tokyo.

Using Streaming

To get streaming responses from your agent:

import asyncio
from python_a2a import HTTPClient

async def stream_example():
    client = HTTPClient("http://localhost:5000")
    
    print("Streaming response:")
    async for chunk in client.stream_response("Paris"):
        print(chunk.content, end="", flush=True)
    print("\nStreaming complete!")

if __name__ == "__main__":
    asyncio.run(stream_example())

Save this as streaming_client.py and run it:

python streaming_client.py

Next Steps

Now that you've created your first A2A agent and client, you can:

1. Create a network of agents
2. Add MCP tools to your agent
3. Integrate with LangChain
4. Explore more examples
5. Learn about advanced features

For more detailed information, check out the Core Concepts section.