A2A Python SDK

Overview

The A2A Python SDK provides a Python client library for building agents that interact with A2A agents following the Agent2Agent (A2A) Protocol. Use this SDK to build agents that:

Discover and connect to StackOne’s A2A agents
Send messages and receive responses
Manage multi-turn conversations with context

Installation

Install the A2A SDK using uv or pip:

uv add a2a-sdk

Quick Start

Get an Agent Card, send a Message, and print the response.

import base64
import uuid
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    Part,
    Message,
    MessageSendParams,
    SendMessageRequest,
    SendMessageResponse,
)

BASE64_ENCODED_STACKONE_API_KEY: str = base64.b64encode(b"<stackone_api_key>:").decode() 
STACKONE_ACCOUNT_ID: str = "<account_id>"

async def main() -> None:
    """
    Fetch an Agent Card, send a message, and print the response.
    """
    async with httpx.AsyncClient(
        headers={
            "Authorization": f"Basic {BASE64_ENCODED_STACKONE_API_KEY}",
            "x-account-id": STACKONE_ACCOUNT_ID
        }
    ) as http_client:
        # Fetch the Agent Card
        card_resolver: A2ACardResolver = A2ACardResolver(
            http_client,
            "https://a2a.stackone.com"
        )
        
        agent_card: AgentCard = await card_resolver.get_agent_card()
        print(agent_card.model_dump_json(indent=4))

        # Initialize client
        client: A2AClient = await A2AClient(http_client, agent_card)

        # Send a message
        message_request: SendMessageRequest = SendMessageRequest(
            params=MessageSendParams(
                message=Message(
                    message_id=str(uuid.uuid4()),
                    parts=[Part(type="text", text="List the first 5 employees")],   
                    role="user",
                )
            )
        )

        response: SendMessageResponse = await client.send_message(message_request)
        print(response.model_dump_json(indent=4))

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

See Authentication Guide for details on obtaining your API key and account ID.

Getting Agent Cards

During agent initialisation, fetch an Agent Card for every A2A agent you want to send messages to. This is typically a setup step as opposed to a tool.

Ensure to include the agent information in the agent’s context (e.g. system message).

import base64
import json
from typing import Any

import httpx
from a2a.client import A2ACardResolver
from a2a.types import AgentCard

BASE64_ENCODED_STACKONE_API_KEY: str = base64.b64encode(b"<stackone_api_key>:").decode()

async def fetch_agent_cards(account_ids: list[str]) -> dict[str, AgentCard]:
    """
    Fetch Agent Cards from StackOne A2A agents for multiple accounts.

    The Agent URL is fixed to https://a2a.stackone.com and dynamically loads
    an Agent Card based on the account ID.

    Args:
        account_ids (list[str]): List of StackOne account IDs

    Returns:
        dict[str, AgentCard]: Dictionary mapping account IDs to their Agent Cards
    """
    agent_cards: dict[str, AgentCard] = {}

    for account_id in account_ids:
        try:
            async with httpx.AsyncClient(
                headers={
                    "Authorization": f"Basic {BASE64_ENCODED_STACKONE_API_KEY}",
                    "x-account-id": account_id
                }
            ) as http_client:
                card_resolver: A2ACardResolver = A2ACardResolver(
                    http_client,
                    "https://a2a.stackone.com"
                )
                card: AgentCard = await card_resolver.get_agent_card()
                agent_cards[card.name] = card

        except Exception as e:
            print(f"Failed to fetch agent card for account {account_id}: {e}")

    return agent_cards


async def get_a2a_agent_info(agent_cards: dict[str, AgentCard]) -> str:
    """
    Get A2A agent information for agent context.

    Args:
        agent_cards (dict[str, AgentCard]): Dictionary mapping account IDs to their Agent Cards

    Returns:
        str: Agent information for LLM context
    """
    a2a_agent_info: list[dict[str, Any]] = []
    
    for agent_card in agent_cards.values():
        a2a_agent_info.append({
            "name": agent_card.name,
            "description": agent_card.description,
            "skills": [
                skill.name for skill in (agent_card.skills or [])
            ]
        })
    
    return "\n".join([
        "<agents>",
        json.dumps(a2a_agent_info, indent=4),
        "</agents>"
    ])


async def main() -> None:
    """
    Fetch agent cards from multiple StackOne accounts and format for LLM context.
    """
    # Fetch agent cards from multiple StackOne accounts
    account_ids: list[str] = ["<account_id_1>", "<account_id_2>", "<account_id_3>"]
    agent_cards: dict[str, AgentCard] = await fetch_agent_cards(account_ids)
    
    # Get agent information for LLM context (e.g. system message)
    a2a_agent_info: str = await get_a2a_agent_info(agent_cards)
    print(a2a_agent_info)

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

Sending Messages

To send messages to A2A agents, give your agent access to a tool like below.

The tool below gives the agent relatively high autonomy. You might want to abstract context IDs and task IDs outside of the tool.

import uuid
from typing import Any, List, Tuple
import base64
import httpx

from a2a.client import A2AClient
from a2a.types import (
    AgentCard,
    JSONRPCErrorResponse,
    Message,
    MessageSendParams,
    Part,
    SendMessageRequest,
    SendMessageResponse,
    Task,
)

BASE64_ENCODED_STACKONE_API_KEY: str = base64.b64encode(b"<stackone_api_key>:").decode() 
STACKONE_ACCOUNT_ID: str = "<account_id>"

# Dictionary mapping account IDs to their Agent Cards
agent_cards: dict[str, AgentCard] = {}

async def send_message_to_agent(
    agent_name: str,
    message: str,
    context_id: str | None = None,
    task_id: str | None = None
) -> dict[str, Any]:
    """
    Send a message to an agent and receive a response. The response can be a Task, Message, or Error.
    
    To start a new conversation, set `context_id` and `task_id` to None.
    
    To continue a conversation, set `context_id` to the `context_id` of the previous conversation.
        - If the previous Task has not terminated (input_required, auth_required), set `task_id` to the `task_id` of the previous Task.
        - If the previous Task has terminated (completed, canceled, rejected, or failed), set `task_id` to None.

    Args:
        agent_name (str): The name of the agent to send the message to
        message (str): The message text to send
        context_id (str | None): The context ID
        task_id (str | None): The task ID

    Returns:
        dict[str, Any]: A dictionary containing the response from the agent
    """
    agent_card: List[Tuple[str, AgentCard]] = [(account_id, agent_card) for account_id, agent_card in agent_cards.items() if agent_card.name == agent_name]
    
    if not len(agent_card):
        raise ValueError(f"Agent {agent_name} not found")

    account_id: str
    agent_card: AgentCard
    account_id, agent_card = agent_card[0]

    with httpx.AsyncClient(
        headers={
            "Authorization": f"Basic {BASE64_ENCODED_STACKONE_API_KEY}",
            "x-account-id": account_id
        }
    ) as http_client:
        client: A2AClient = await A2AClient(http_client, agent_card)
        
    send_message_request: SendMessageRequest = SendMessageRequest(
        params=MessageSendParams(
            message=Message(
                message_id=str(uuid.uuid4()),
                parts=[Part(type="text", text=message)],
                role="user",
            ),
            context_id=context_id,
            task_id=task_id
        )
    )

    response: SendMessageResponse = await client.send_message(send_message_request)

    # Handle error
    if isinstance(response.root, JSONRPCErrorResponse):
        return response.root

    result: Message | Task = response.root.result

    if isinstance(result, Message):
        # Agent responded with a Message (i.e. didn't start a Task)
        # Return the Message
        return {
            "Message": result.model_dump()
        }
    else:
        # Agent responded with a Task (i.e. started a Task)
        # Tasks contain the whole message history. We typically want the
        # Task's status, and Artifacts (if the Task has completed).
        return {
            "Task": {
                "status": result.status.model_dump(),
                "artifacts": [artifact.model_dump() for artifact in result.artifacts],
                "context_id": result.context_id,
                "task_id": result.id,
            }
        }

Next Steps

JavaScript SDK

Use the A2A JavaScript SDK

Agent Frameworks

Build agents with framework integrations

Authentication

Learn about authentication and security

Quickstart

Learn A2A basics with cURL

Getting Started

A2A SDKs

Agent Frameworks

Overview

Installation

Quick Start

Getting Agent Cards

Sending Messages

Next Steps

JavaScript SDK

Agent Frameworks

Authentication

Quickstart

Getting Started

A2A SDKs

Agent Frameworks

​Overview

​Installation

​Quick Start

​Getting Agent Cards

​Sending Messages

​Next Steps

JavaScript SDK

Agent Frameworks

Authentication

Quickstart

Overview

Installation

Quick Start

Getting Agent Cards

Sending Messages

Next Steps