A2A Examples

This page provides detailed examples of using the Agent-to-Agent (A2A) protocol with Scoras. These examples demonstrate how to create A2A servers, connect to them with clients, and build multi-agent systems using the A2A protocol.

Basic A2A Server Example

This example shows how to create a simple A2A server with an agent:

import asyncio
import json
import uuid
from typing import Dict, Any, List

import scoras as sc
from scoras.a2a import create_agent_skill, create_a2a_server, run_a2a_server

# Define skills for our A2A agent
math_skill = create_agent_skill(
    id="math",
    name="Mathematics",
    description="Perform mathematical calculations and solve problems",
    tags=["math", "calculation", "problem-solving"],
    examples=[
        "Calculate the derivative of f(x) = x^2 + 3x + 2",
        "Solve the equation 2x + 5 = 13"
    ],
    complexity="standard"
)

research_skill = create_agent_skill(
    id="research",
    name="Research",
    description="Find and analyze information on various topics",
    tags=["research", "information", "analysis"],
    examples=[
        "Summarize the latest research on renewable energy",
        "Compare and contrast different machine learning algorithms"
    ],
    complexity="complex"
)

async def main():
    # Create an agent
    agent = sc.Agent(
        model="openai:gpt-4o",
        system_prompt="You are a helpful assistant with expertise in mathematics and research.",
        enable_scoring=True
    )

    # Create an A2A server
    server = create_a2a_server(
        name="ScorasAgent",
        description="A versatile agent with multiple skills powered by Scoras",
        agent=agent,
        skills=[math_skill, research_skill],
        provider={
            "organization": "Scoras Project",
            "url": "https://scoras.example.com"
        },
        capabilities={
            "streaming": True,
            "push_notifications": False,
            "state_transition_history": True
        },
        authentication_schemes=["bearer"],
        enable_scoring=True
    )

    # Print the agent card
    agent_card = server.get_agent_card()
    print("Agent Card:", json.dumps(agent_card.model_dump(), indent=2))

    # Simulate handling a task
    print("\nSimulating handling a task...")
    task_id = str(uuid.uuid4())
    task_data = {
        "id": task_id,
        "messages": [
            {
                "role": "user",
                "parts": [
                    {
                        "type": "text",
                        "text": "Calculate the area of a circle with radius 5 cm."
                    }
                ]
            }
        ]
    }

    # Handle the task
    response = await server.handle_request({
        "jsonrpc": "2.0",
        "method": "tasks/send",
        "params": {
            "task": task_data
        },
        "id": "request1"
    })

    print("Response:", json.dumps(response, indent=2))

    # Get the complexity score
    score = server.get_complexity_score()
    print("\nServer Complexity Score:", json.dumps(score, indent=2))

    # Run the server
    print("\nStarting A2A server on http://0.0.0.0:8001")
    await run_a2a_server(server, host="0.0.0.0", port=8001)

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

A2A Client Example

This example shows how to connect to an A2A server and use its capabilities:

import asyncio
import json
import uuid
from typing import Dict, Any, List

import scoras as sc
from scoras.a2a import A2AClient

async def main():
    # Create an A2A client
    client = A2AClient(
        agent_url="http://localhost:8001",
        enable_scoring=True
    )

    # Get the agent card
    agent_card = await client.get_agent_card()
    print("Agent Card:")
    print(f"  Name: {agent_card.name}")
    print(f"  Description: {agent_card.description}")
    print(f"  Skills: {[skill.name for skill in agent_card.skills]}")

    # Send a task to the agent
    print("\nSending a math task to the agent...")
    math_task = await client.send_task(
        message="Calculate the area of a circle with radius 5 cm."
    )
    print(f"Task ID: {math_task.id}")
    print(f"Task State: {math_task.state}")

    # Wait for the task to complete
    print("\nWaiting for task to complete...")
    math_task = await client.wait_for_task(math_task.id)

    # Print the response
    print("\nTask completed!")
    print(f"Response: {math_task.messages[-1].parts[0].text}")

    # Send a research task
    print("\nSending a research task to the agent...")
    research_task = await client.send_task(
        message="Summarize the key concepts of quantum computing."
    )

    # Wait for the task to complete
    print("\nWaiting for task to complete...")
    research_task = await client.wait_for_task(research_task.id)

    # Print the response
    print("\nTask completed!")
    print(f"Response: {research_task.messages[-1].parts[0].text}")

    # Get the complexity score
    score = client.get_complexity_score()
    print("\nClient Complexity Score:", json.dumps(score, indent=2))

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

A2A Task Management Example

This example demonstrates managing tasks with the A2A protocol:

import asyncio
import json
import uuid
from typing import Dict, Any, List

import scoras as sc
from scoras.a2a import A2AClient

async def main():
    # Create an A2A client
    client = A2AClient(
        agent_url="http://localhost:8001",
        enable_scoring=True
    )

    # Create a new task
    print("Creating a new task...")
    task = await client.send_task(
        message="What is the formula for calculating the area of a circle?"
    )
    print(f"Task created with ID: {task.id}")
    print(f"Initial state: {task.state}")

    # Get the task
    print("\nRetrieving task...")
    retrieved_task = await client.get_task(task.id)
    print(f"Retrieved task state: {retrieved_task.state}")

    # Send a follow-up message
    print("\nSending follow-up message...")
    updated_task = await client.send_message(
        task_id=task.id,
        message="Also, what is the formula for the circumference?"
    )
    print(f"Updated task state: {updated_task.state}")

    # Wait for the task to complete
    print("\nWaiting for task to complete...")
    completed_task = await client.wait_for_task(task.id)
    print(f"Final task state: {completed_task.state}")

    # Print the conversation
    print("\nFull conversation:")
    for message in completed_task.messages:
        role = message.role
        text = message.parts[0].text if message.parts else ""
        print(f"{role}: {text}")

    # Cancel a task (demonstration only)
    print("\nDemonstrating task cancellation...")
    new_task = await client.send_task(
        message="This task will be cancelled."
    )
    print(f"Created task with ID: {new_task.id}")

    cancelled_task = await client.cancel_task(new_task.id)
    print(f"Cancelled task state: {cancelled_task.state}")

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

A2A Agent Adapter Example

This example shows how to adapt a Scoras agent to use A2A agents:

import asyncio
import json

import scoras as sc
from scoras.a2a import A2AAgentAdapter

async def main():
    # Create a Scoras agent
    agent = sc.Agent(
        model="openai:gpt-4o",
        system_prompt="You are a helpful assistant that can coordinate with other agents.",
        enable_scoring=True
    )

    # Create an A2A agent adapter
    adapter = A2AAgentAdapter(
        agent=agent,
        enable_scoring=True
    )

    # Connect to an A2A agent
    print("Connecting to A2A agent...")
    await adapter.connect_to_agent("http://localhost:8001")

    # Get the agent card
    agent_card = adapter.get_agent_card()
    print(f"Connected to agent: {agent_card.name}")
    print(f"Available skills: {[skill.name for skill in agent_card.skills]}")

    # Run the agent with a query that will use the math skill
    print("\nRunning agent with math query...")
    math_response = await adapter.run("I need to calculate the area of a circle with radius 5 cm.")
    print("Response:", math_response)

    # Run the agent with a query that will use the research skill
    print("\nRunning agent with research query...")
    research_response = await adapter.run("Can you summarize the key concepts of quantum computing?")
    print("Response:", research_response)

    # Get the complexity score
    score = adapter.get_complexity_score()
    print("\nAdapter Complexity Score:", json.dumps(score, indent=2))

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

Streaming A2A Example

This example demonstrates streaming responses from an A2A agent:

import asyncio
import json

import scoras as sc
from scoras.a2a import A2AClient

async def main():
    # Create an A2A client
    client = A2AClient(
        agent_url="http://localhost:8001",
        enable_scoring=True
    )

    # Stream a response from the agent
    print("\nStreaming response from agent...")
    task_id = None

    async for chunk in client.stream_task(
        message="Explain the concept of quantum entanglement in simple terms."
    ):
        if task_id is None:
            task_id = chunk.get("task_id")
            print(f"Task ID: {task_id}")

        if "delta" in chunk:
            print(chunk["delta"], end="", flush=True)

    print("\n\nStreaming completed!")

    # Get the final task
    if task_id:
        task = await client.get_task(task_id)
        print(f"\nFinal task state: {task.state}")

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

Multi-Agent System Example

This example demonstrates building a multi-agent system with A2A:

import asyncio
import json
from typing import Dict, Any, List

import scoras as sc
from scoras.a2a import A2AClient, create_a2a_server, create_agent_skill, run_a2a_server

# Note: This example assumes you have multiple A2A agents running on different ports

async def setup_math_agent():
    """Set up and run a math specialist agent."""
    math_skill = create_agent_skill(
        id="math",
        name="Mathematics",
        description="Perform mathematical calculations and solve problems",
        complexity="standard"
    )

    math_agent = sc.Agent(
        model="openai:gpt-4o",
        system_prompt="You are a mathematics expert. You excel at solving math problems and explaining mathematical concepts.",
        enable_scoring=True
    )

    math_server = create_a2a_server(
        name="MathAgent",
        description="A specialized agent for mathematics",
        agent=math_agent,
        skills=[math_skill],
        enable_scoring=True
    )

    # Run in background
    asyncio.create_task(run_a2a_server(math_server, host="0.0.0.0", port=8001))
    return math_server

async def setup_research_agent():
    """Set up and run a research specialist agent."""
    research_skill = create_agent_skill(
        id="research",
        name="Research",
        description="Find and analyze information on various topics",
        complexity="complex"
    )

    research_agent = sc.Agent(
        model="anthropic:claude-3-opus",
        system_prompt="You are a research specialist. You excel at finding and analyzing information on various topics.",
        enable_scoring=True
    )

    research_server = create_a2a_server(
        name="ResearchAgent",
        description="A specialized agent for research",
        agent=research_agent,
        skills=[research_skill],
        enable_scoring=True
    )

    # Run in background
    asyncio.create_task(run_a2a_server(research_server, host="0.0.0.0", port=8002))
    return research_server

async def setup_writing_agent():
    """Set up and run a writing specialist agent."""
    writing_skill = create_agent_skill(
        id="writing",
        name="Writing",
        description="Create and edit written content",
        complexity="standard"
    )

    writing_agent = sc.Agent(
        model="gemini:gemini-pro",
        system_prompt="You are a writing expert. You excel at creating clear, engaging, and well-structured content.",
        enable_scoring=True
    )

    writing_server = create_a2a_server(
        name="WritingAgent",
        description="A specialized agent for writing",
        agent=writing_agent,
        skills=[writing_skill],
        enable_scoring=True
    )

    # Run in background
    asyncio.create_task(run_a2a_server(writing_server, host="0.0.0.0", port=8003))
    return writing_server

async def run_coordinator():
    """Run a coordinator agent that orchestrates the specialist agents."""
    # Set up the specialist agents
    await setup_math_agent()
    await setup_research_agent()
    await setup_writing_agent()

    # Give servers time to start
    await asyncio.sleep(2)

    # Connect to the specialist agents
    math_client = A2AClient(agent_url="http://localhost:8001")
    research_client = A2AClient(agent_url="http://localhost:8002")
    writing_client = A2AClient(agent_url="http://localhost:8003")

    # Create a coordinator agent
    coordinator = sc.Agent(
        model="openai:gpt-4o",
        system_prompt="You are a coordinator agent that delegates tasks to specialist agents and combines their results.",
        enable_scoring=True
    )

    # Process a complex query
    query = "Create a 500-word article about quantum computing, including mathematical explanations of quantum bits and superposition."
    print(f"Processing query: {query}")

    # Step 1: Research quantum computing
    print("\nStep 1: Delegating research task...")
    research_task = await research_client.send_task(
        message="Research quantum computing, focusing on key concepts and recent developments."
    )
    research_task = await research_client.wait_for_task(research_task.id)
    research_result = research_task.messages[-1].parts[0].text
    print("Research completed!")

    # Step 2: Get mathematical explanations
    print("\nStep 2: Delegating math task...")
    math_task = await math_client.send_task(
        message="Explain quantum bits and superposition mathematically. Include relevant equations and concepts."
    )
    math_task = await math_client.wait_for_task(math_task.id)
    math_result = math_task.messages[-1].parts[0].text
    print("Math explanation completed!")

    # Step 3: Write the article
    print("\nStep 3: Delegating writing task...")
    writing_prompt = f"""
    Create a 500-word article about quantum computing based on the following research and mathematical explanations:

    RESEARCH:
    {research_result}

    MATHEMATICAL EXPLANATIONS:
    {math_result}

    The article should be engaging, clear, and accessible to a general audience while still including the mathematical concepts.
    """

    writing_task = await writing_client.send_task(message=writing_prompt)
    writing_task = await writing_client.wait_for_task(writing_task.id)
    final_article = writing_task.messages[-1].parts[0].text
    print("Article writing completed!")

    # Print the final result
    print("\n=== FINAL ARTICLE ===\n")
    print(final_article)

    # Get complexity scores
    math_score = await math_client.get_complexity_score()
    research_score = await research_client.get_complexity_score()
    writing_score = await writing_client.get_complexity_score()

    print("\nComplexity Scores:")
    print(f"Math Agent: {math_score['complexity_rating']} (Score: {math_score['total_score']})")
    print(f"Research Agent: {research_score['complexity_rating']} (Score: {research_score['total_score']})")
    print(f"Writing Agent: {writing_score['complexity_rating']} (Score: {writing_score['total_score']})")

    # Calculate total system complexity
    total_score = math_score['total_score'] + research_score['total_score'] + writing_score['total_score']
    print(f"Total System Complexity Score: {total_score}")

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

A2A with Custom Authentication Example

This example demonstrates using custom authentication with A2A:

import asyncio
import json
import uuid
from typing import Dict, Any, List, Optional

import scoras as sc
from scoras.a2a import create_a2a_server, run_a2a_server, A2AClient

# Custom authentication handler
class CustomAuthHandler:
    def __init__(self):
        self.api_keys = {
            "valid-api-key": "user1",
            "another-valid-key": "user2"
        }

    async def authenticate(self, auth_header: Optional[str]) -> Optional[Dict[str, Any]]:
        """Authenticate a request based on the Authorization header."""
        if not auth_header or not auth_header.startswith("Bearer "):
            return None

        api_key = auth_header[7:]  # Remove "Bearer " prefix

        if api_key in self.api_keys:
            # Return user info on successful authentication
            return {
                "user_id": self.api_keys[api_key],
                "authenticated": True
            }

        return None

async def main():
    # Create an agent
    agent = sc.Agent(
        model="openai:gpt-4o",
        system_prompt="You are a helpful assistant.",
        enable_scoring=True
    )

    # Create skills
    math_skill = create_agent_skill(
        id="math",
        name="Mathematics",
        description="Perform mathematical calculations",
        complexity="standard"
    )

    # Create auth handler
    auth_handler = CustomAuthHandler()

    # Create an A2A server with custom authentication
    server = create_a2a_server(
        name="SecureAgent",
        description="A secure agent with custom authentication",
        agent=agent,
        skills=[math_skill],
        authentication_schemes=["bearer"],
        authentication_handler=auth_handler.authenticate,
        enable_scoring=True
    )

    # Run the server in the background
    asyncio.create_task(run_a2a_server(server, host="0.0.0.0", port=8001))

    # Give the server time to start
    await asyncio.sleep(1)

    # Create an A2A client with authentication
    client = A2AClient(
        agent_url="http://localhost:8001",
        auth_token="valid-api-key",
        enable_scoring=True
    )

    # Send a task to the agent
    print("Sending task with valid authentication...")
    try:
        task = await client.send_task(
            message="Calculate the area of a circle with radius 5 cm."
        )
        print(f"Task created with ID: {task.id}")

        # Wait for the task to complete
        completed_task = await client.wait_for_task(task.id)
        print(f"Response: {completed_task.messages[-1].parts[0].text}")
    except Exception as e:
        print(f"Error: {e}")

    # Create a client with invalid authentication
    invalid_client = A2AClient(
        agent_url="http://localhost:8001",
        auth_token="invalid-api-key",
        enable_scoring=True
    )

    # Try to send a task with invalid authentication
    print("\nSending task with invalid authentication...")
    try:
        task = await invalid_client.send_task(
            message="This should fail due to invalid authentication."
        )
        print(f"Task created with ID: {task.id}")
    except Exception as e:
        print(f"Error as expected: {e}")

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

Next Steps

• Check out the MCP Examples for model-tool communication
• Learn about A2A Protocol for more details
• Explore the A2A API Reference for comprehensive documentation