A2A Python SDK

import asyncio
import base64
import uuid
import httpx
from a2a.client import A2ACardResolver, ClientFactory, ClientConfig
from a2a.types import Message, Part, TextPart

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(http_client, "https://a2a.stackone.com")
        card = await card_resolver.get_agent_card()
        print(card.model_dump_json(indent=4))

        # Initialize client
        factory = ClientFactory(ClientConfig(httpx_client=http_client))
        client = factory.create(card)

        # Send a message
        message = Message(
            messageId=str(uuid.uuid4()),
            role="user",
            parts=[Part(root=TextPart(kind="text", text="List the first 5 employees"))],
            kind="message"
        )

        # send_message returns an async generator of streaming events
        async for event in client.send_message(message):
            if isinstance(event, tuple):
                task, update = event
                if task.status.state == "completed":
                    for artifact in (task.artifacts or []):
                        for part in (artifact.parts or []):
                            if hasattr(part.root, "text"):
                                print(part.root.text)

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

import asyncio
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(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__":
    asyncio.run(main())

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

from a2a.client import A2ACardResolver, ClientFactory, ClientConfig
from a2a.types import (
    AgentCard,
    Message,
    Part,
    TextPart,
    Task,
)

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

# Dictionary mapping agent names to their (account_id, AgentCard) pairs
agent_cards: dict[str, tuple[str, AgentCard]] = {}


async def send_message_to_agent(
    agent_name: str,
    message: str,
) -> dict[str, Any]:
    """
    Send a message to an agent and receive a response.

    Args:
        agent_name (str): The name of the agent to send the message to
        message (str): The message text to send

    Returns:
        dict[str, Any]: A dictionary containing the response from the agent
    """
    if agent_name not in agent_cards:
        raise ValueError(f"Agent {agent_name} not found")

    account_id, card = agent_cards[agent_name]

    async with httpx.AsyncClient(
        headers={
            "Authorization": f"Basic {BASE64_ENCODED_STACKONE_API_KEY}",
            "x-account-id": account_id
        }
    ) as http_client:
        factory = ClientFactory(ClientConfig(httpx_client=http_client))
        client = factory.create(card)

        msg = Message(
            messageId=str(uuid.uuid4()),
            role="user",
            parts=[Part(root=TextPart(kind="text", text=message))],
            kind="message"
        )

        final_task: Task | None = None
        async for event in client.send_message(msg):
            if isinstance(event, tuple):
                task, update = event
                final_task = task

        if final_task is None:
            return {"error": "No response received"}

        # Tasks contain the whole message history. We typically want the
        # Task's status, and Artifacts (if the Task has completed).
        return {
            "Task": {
                "status": final_task.status.model_dump(),
                "artifacts": [artifact.model_dump() for artifact in (final_task.artifacts or [])],
                "context_id": final_task.contextId,
                "task_id": final_task.id,
            }
        }