Skip to content

Core Concepts

Jump to bottom
Manoj Desai edited this page May 3, 2025 · 1 revision

Core Concepts

Python A2A is built on a set of core concepts derived from the Agent-to-Agent (A2A) protocol and extended with additional features.

A2A Protocol

The A2A protocol defines standardized APIs for agent-to-agent communication, enabling interoperability between different agent implementations. The protocol consists of:

1. Agents

Agents are the primary actors in the A2A ecosystem. Each agent:

  • Has a unique identifier and metadata
  • Exposes a well-defined API for communication
  • Can process tasks and messages
  • May have specialized skills or capabilities
@agent(
    name="Research Agent",
    description="Performs research on various topics",
    version="1.0.0"
)
class ResearchAgent(A2AServer):
    # Agent implementation

2. Tasks

Tasks are discrete units of work that can be assigned to agents. Each task:

  • Has a unique identifier
  • Contains input data
  • Can have a status (pending, running, completed, etc.)
  • Returns output upon completion
# Creating a task
task = client.create_task({"query": "Research quantum computing advances"})

# Getting task status
status = client.get_task_status(task.task_id)

# Getting task output
output = client.get_task_output(task.task_id)

3. Messages

Messages represent communication between agents or from users to agents:

  • Can contain text, structured data, or attachments
  • Support rich content formatting
  • Can be sent synchronously or streamed
# Sending a message
response = client.send_message("What's the weather in New York?")

# Streaming a message
async for chunk in client.stream_response("Tell me a long story"):
    print(chunk.content, end="", flush=True)

Python A2A Extensions

Python A2A extends the base A2A protocol with additional features:

1. Skills

Skills are specialized capabilities that agents can expose:

  • Documented with metadata for discovery
  • Can be invoked directly or through natural language
  • Help organize agent functionality
@skill(
    name="Weather Lookup",
    description="Get weather for a location",
    parameters={"location": {"type": "string", "description": "City name"}}
)
def get_weather(self, location):
    # Implementation

2. Agent Cards

Agent cards provide rich metadata about agents:

  • Name, description, and version
  • Available skills and capabilities
  • Usage examples and documentation
  • Contact information
agent_card = AgentCard(
    name="Travel Assistant",
    description="Helps plan travel itineraries",
    skills=[SkillCard(...), SkillCard(...)],
    examples=["Plan a trip to Paris", "Find hotels in Tokyo"]
)

3. Conversations

Conversations represent an ongoing exchange between a user and an agent:

  • Maintain context across multiple messages
  • Can be persisted for later continuation
  • Support multi-turn interactions
# Starting a conversation
conversation = client.create_conversation()

# Adding to the conversation
response = client.add_message(conversation.id, "Hello, agent!")
response = client.add_message(conversation.id, "Tell me more about that.")

4. Streaming

Streaming enables real-time delivery of agent responses:

  • Provides immediate feedback for long-running operations
  • Supports token-by-token delivery for LLM outputs
  • Uses industry-standard server-sent events (SSE)
async for chunk in client.stream_response("Generate a detailed report"):
    print(chunk.content, end="", flush=True)

MCP Integration

The Model Context Protocol (MCP) provides a standardized way for agents to access tools, data sources, and external systems:

1. Tools

Tools are functions that can be exposed via MCP:

  • Have well-defined schemas (parameters and return types)
  • Can be invoked synchronously or asynchronously
  • Support validation and error handling
@mcp_agent.tool()
def search_database(query: str) -> List[Dict[str, Any]]:
    """Search the database for information."""
    # Implementation

2. MCP Agents

MCP agents can use tools to extend their capabilities:

  • Can expose multiple tools
  • Handle tool selection based on user queries
  • Process tool results in responses
mcp_agent = MCPAgent(
    name="Research Assistant",
    description="Research assistant with database access"
)

@mcp_agent.tool()
def search_database(query: str) -> List[Dict[str, Any]]:
    # Implementation

Agent Networks

Agent networks enable the coordination of multiple specialized agents:

  • Compose multiple agents into a unified system
  • Route requests to appropriate specialists
  • Enable agent collaboration on complex tasks
network = AgentNetwork(name="Travel Assistant Network")
network.add("weather", "http://localhost:5001")
network.add("hotels", "http://localhost:5002")
network.add("flights", "http://localhost:5003")

Workflows

Workflows orchestrate complex interactions between agents:

  • Define sequential or conditional steps
  • Support branching based on agent responses
  • Enable complex multi-agent collaboration
flow = Flow(agent_network=network)
flow.ask("research", "Research tourist attractions in {destination}")
flow.if_contains("beach")
flow.ask("beach_expert", "Recommend beach activities in {destination}")
flow.else_if_contains("mountains")
flow.ask("hiking_expert", "Suggest hiking trails in {destination}")

These core concepts form the foundation of the Python A2A library and provide the building blocks for creating sophisticated agent ecosystems.

Clone this wiki locally