A2A SDK

# uv (or: pip install 'a2a-sdk>=0.3,<0.4')
uv add 'a2a-sdk>=0.3,<0.4'

npm install @a2a-js/sdk@^0.3

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 { A2AClient, SendMessageSuccessResponse } from '@a2a-js/sdk/client';
import { Message, MessageSendParams } from '@a2a-js/sdk';
import { v4 as uuidv4 } from 'uuid';

const STACKONE_API_KEY = '<stackone_api_key>';
const STACKONE_ACCOUNT_ID = '<account_id>';

// Base64 encode the API key (with trailing colon for Basic auth)
const base64EncodedApiKey = Buffer.from(`${STACKONE_API_KEY}:`).toString('base64');

async function main() {
  // Create a custom fetch that adds authentication headers
  const authenticatedFetch: typeof fetch = async (url, init) => {
    const headers = new Headers(init?.headers);
    headers.set('Authorization', `Basic ${base64EncodedApiKey}`);
    headers.set('x-account-id', STACKONE_ACCOUNT_ID);
    return fetch(url, { ...init, headers });
  };

  // Create a client pointing to the StackOne A2A Agent Card URL
  const client = await A2AClient.fromCardUrl(
    'https://a2a.stackone.com/.well-known/agent-card.json',
    { fetchImpl: authenticatedFetch }
  );

  // Send a message
  const sendParams: MessageSendParams = {
    message: {
      messageId: uuidv4(),
      role: 'user',
      parts: [{ kind: 'text', text: 'List the first 5 employees' }],
      kind: 'message',
    },
  };

  const response = await client.sendMessage(sendParams);

  if ('error' in response) {
    console.error('Error:', response.error.message);
  } else {
    const result = (response as SendMessageSuccessResponse).result;
    console.log('Agent response:', JSON.stringify(result, null, 2));
  }
}

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 { A2AClient } from '@a2a-js/sdk/client';
import { AgentCard } from '@a2a-js/sdk';

const STACKONE_API_KEY = '<stackone_api_key>';
const base64EncodedApiKey = Buffer.from(`${STACKONE_API_KEY}:`).toString('base64');

interface AgentCardWithAccountId {
  accountId: string;
  agentCard: 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.
 */
async function fetchAgentCards(accountIds: string[]): Promise<Map<string, AgentCardWithAccountId>> {
  const agentCards = new Map<string, AgentCardWithAccountId>();

  for (const accountId of accountIds) {
    try {
      // Create authenticated fetch for this account
      const authenticatedFetch: typeof fetch = async (url, init) => {
        const headers = new Headers(init?.headers);
        headers.set('Authorization', `Basic ${base64EncodedApiKey}`);
        headers.set('x-account-id', accountId);
        return fetch(url, { ...init, headers });
      };

      const client = await A2AClient.fromCardUrl(
        'https://a2a.stackone.com/.well-known/agent-card.json',
        { fetchImpl: authenticatedFetch }
      );

      const agentCard = client.agentCard;
      agentCards.set(agentCard.name, { accountId, agentCard });
    } catch (error) {
      console.error(`Failed to fetch agent card for account ${accountId}:`, error);
    }
  }

  return agentCards;
}

/**
 * Get A2A agent information for agent context.
 */
function getA2AAgentInfo(agentCards: Map<string, AgentCardWithAccountId>): string {
  const a2aAgentInfo: Array<{ name: string; description: string; skills: string[] }> = [];

  for (const { agentCard } of agentCards.values()) {
    a2aAgentInfo.push({
      name: agentCard.name,
      description: agentCard.description,
      skills: (agentCard.skills || []).map((skill) => skill.name),
    });
  }

  return ['<agents>', JSON.stringify(a2aAgentInfo, null, 4), '</agents>'].join('\n');
}

async function main() {
  // Fetch agent cards from multiple StackOne accounts
  const accountIds = ['<account_id_1>', '<account_id_2>', '<account_id_3>'];
  const agentCards = await fetchAgentCards(accountIds);

  // Get agent information for LLM context (e.g. system message)
  const a2aAgentInfo = getA2AAgentInfo(agentCards);
  console.log(a2aAgentInfo);
}

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,
            }
        }

import { A2AClient, SendMessageSuccessResponse } from '@a2a-js/sdk/client';
import { AgentCard, Message, MessageSendParams, Task } from '@a2a-js/sdk';
import { v4 as uuidv4 } from 'uuid';

const STACKONE_API_KEY = '<stackone_api_key>';
const base64EncodedApiKey = Buffer.from(`${STACKONE_API_KEY}:`).toString('base64');

interface AgentCardWithAccountId {
  accountId: string;
  agentCard: AgentCard;
}

// Map of agent names to their cards and account IDs
const agentCards = new Map<string, AgentCardWithAccountId>();

interface SendMessageResult {
  Message?: {
    messageId: string;
    role: string;
    parts: Array<{ kind: string; text?: string }>;
    contextId?: string;
  };
  Task?: {
    status: { state: string; timestamp: string };
    artifacts: Array<{ artifactId: string; name?: string; parts: Array<{ kind: string; text?: string }> }>;
    contextId: string;
    taskId: string;
  };
  error?: { code: number; message: string };
}

/**
 * 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 `contextId` and `taskId` to undefined.
 *
 * To continue a conversation, set `contextId` to the `contextId` of the previous conversation.
 *   - If the previous Task has not terminated (input_required, auth_required),
 *     set `taskId` to the `taskId` of the previous Task.
 *   - If the previous Task has terminated (completed, canceled, rejected, or failed),
 *     set `taskId` to undefined.
 */
async function sendMessageToAgent(
  agentName: string,
  message: string,
  contextId?: string,
  taskId?: string
): Promise<SendMessageResult> {
  const agentEntry = agentCards.get(agentName);

  if (!agentEntry) {
    throw new Error(`Agent ${agentName} not found`);
  }

  const { accountId, agentCard } = agentEntry;

  // Create authenticated fetch for this account
  const authenticatedFetch: typeof fetch = async (url, init) => {
    const headers = new Headers(init?.headers);
    headers.set('Authorization', `Basic ${base64EncodedApiKey}`);
    headers.set('x-account-id', accountId);
    return fetch(url, { ...init, headers });
  };

  const client = await A2AClient.fromCardUrl(agentCard.url, {
    fetchImpl: authenticatedFetch,
  });

  const sendParams: MessageSendParams = {
    message: {
      messageId: uuidv4(),
      role: 'user',
      parts: [{ kind: 'text', text: message }],
      kind: 'message',
    },
    contextId,
    taskId,
  };

  const response = await client.sendMessage(sendParams);

  // Handle error
  if ('error' in response) {
    return { error: response.error };
  }

  const result = (response as SendMessageSuccessResponse).result;

  if (result.kind === 'message') {
    // Agent responded with a Message (i.e. didn't start a Task)
    const msg = result as Message;
    return {
      Message: {
        messageId: msg.messageId,
        role: msg.role,
        parts: msg.parts,
        contextId: msg.contextId,
      },
    };
  } 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).
    const task = result as Task;
    return {
      Task: {
        status: task.status,
        artifacts: task.artifacts || [],
        contextId: task.contextId,
        taskId: task.id,
      },
    };
  }
}