A2A Module

Access external A2A-compatible agents through the AgentVisor A2A Gateway.

Overview

AgentVisor provides two A2A modules:

Module	Use Case
langchain	LangChain/LangGraph integration (recommended)
a2a	Low-level API for direct control

All A2A operations are policy-checked by the host runtime. See A2A Gateway Guide for architecture details.

LangChain Integration

The recommended way to use A2A agents is via the LangChain adapter:

from agentvisor.langchain.a2a import get_a2a_tools

get_a2a_tools()

def get_a2a_tools(agent_name: str | None = None) -> list[StructuredTool]:
    """Get LangChain tools for A2A agents.

    Args:
        agent_name: Optional specific agent to get tools for.
                    If None, returns one tool per configured agent.

    Returns:
        List of LangChain StructuredTool instances.
    """

Example: ReAct Agent with A2A Tools

from agentvisor.langchain.a2a import get_a2a_tools
from langgraph.prebuilt import create_react_agent
from langchain_ollama import ChatOllama

# Get tools for all configured A2A agents
tools = get_a2a_tools()

# Create agent with A2A tools
model = ChatOllama(model="llama3.2")
graph = create_react_agent(model, tools)

Example: Specific Agent Skills

# Get tools for a specific agent's skills
research_tools = get_a2a_tools("research-agent")
code_tools = get_a2a_tools("code-agent")

# Combine tools
all_tools = research_tools + code_tools

Tool Naming

Tools are named based on the agent and skill:

# For generic agent tools (no specific agent name):
tools = get_a2a_tools()
# a2a_research-agent
# a2a_code-agent

# For specific agent skills:
tools = get_a2a_tools("research-agent")
# research-agent_summarize
# research-agent_search

Low-Level API

For direct control over A2A operations:

from agentvisor import a2a

Data Classes

The A2A module provides the following data classes:

AgentInfo

@dataclass
class AgentInfo:
    """Information about an available A2A agent."""
    name: str           # Unique agent identifier
    url: str            # Agent endpoint URL
    description: str    # Human-readable description
    skills: list[Skill] # Available skills

Skill

@dataclass
class Skill:
    """A skill (capability) offered by an A2A agent."""
    id: str                    # Skill identifier
    name: str                  # Human-readable name
    description: str           # Skill description
    input_schema: dict[str, Any]  # JSON Schema for input
    examples: list[str]        # Example prompts/inputs for this skill

AgentCard

@dataclass
class AgentCard:
    """Full capability description of an A2A agent."""
    name: str               # Agent name
    description: str        # Agent description
    url: str                # Agent endpoint URL
    version: str            # Protocol version
    capabilities: Capabilities
    skills: list[Skill]

Capabilities

@dataclass
class Capabilities:
    """Agent capabilities."""
    streaming: bool          # Supports SSE streaming
    push_notifications: bool # Supports push notifications

Message

@dataclass
class Message:
    """An A2A message."""
    role: str             # "user" or "agent"
    parts: list[Part]     # Message content
    message_id: str | None
    context_id: str | None
    task_id: str | None
    metadata: dict[str, Any] | None

Part

@dataclass
class Part:
    """A message content part."""
    text: str | None
    data: bytes | None
    file: FilePart | None
    metadata: dict[str, Any] | None

FilePart

@dataclass
class FilePart:
    """A file attachment in an A2A message."""
    name: str | None
    mime_type: str | None
    bytes: bytes | None
    uri: str | None

Task

@dataclass
class Task:
    """An A2A task (unit of work)."""
    id: str
    context_id: str | None
    status: TaskStatus
    history: list[Message]     # Conversation history
    artifacts: list[Artifact]  # Output artifacts
    metadata: dict[str, Any] | None

TaskStatus

@dataclass
class TaskStatus:
    """Current task state."""
    state: str            # working, input_required, auth_required, completed, failed, canceled, rejected
    message: Message | None
    timestamp: str | None

Artifact

@dataclass
class Artifact:
    """Output artifact from a task."""
    name: str | None
    description: str | None
    parts: list[Part]
    index: int
    metadata: dict[str, Any] | None

TaskUpdate

@dataclass
class TaskUpdate:
    """A task update event."""
    task: Task
    event_type: str

list_agents()

def list_agents() -> list[AgentInfo]:
    """List available A2A agents configured in the gateway.

    Returns:
        List of AgentInfo objects describing available agents.
    """

Example

from agentvisor import a2a

agents = a2a.list_agents()
for agent in agents:
    print(f"{agent.name}: {agent.description}")
    for skill in agent.skills:
        print(f"  - {skill.name}: {skill.description}")

get_agent_card()

def get_agent_card(agent_name: str) -> AgentCard:
    """Get the Agent Card (capability description) for an agent.

    Args:
        agent_name: The name of the agent to query.

    Returns:
        AgentCard with full capability information.

    Raises:
        AuthorizationError: If access to this agent is denied by policy.
    """

Example

from agentvisor import a2a

card = a2a.get_agent_card("research-agent")
print(f"Version: {card.version}")
print(f"Streaming: {card.capabilities.streaming}")
for skill in card.skills:
    print(f"  - {skill.name}: {skill.description}")

send_task()

def send_task(
    agent_name: str,
    message: Message,
    context_id: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> Task:
    """Send a task to an external A2A agent.

    Args:
        agent_name: The name of the agent to send the task to.
        message: The message to send.
        context_id: Optional context ID to continue a conversation.
        metadata: Optional metadata to include with the request.

    Returns:
        Task object with the task ID and current status.

    Raises:
        AuthorizationError: If this action is denied by policy.
    """

Example

from agentvisor import a2a

# Create a simple text message
message = a2a.text_message("Summarize the quarterly report")

# Send the task
task = a2a.send_task("research-agent", message)
print(f"Task ID: {task.id}")
print(f"Status: {task.status.state}")

get_task()

def get_task(agent_name: str, task_id: str) -> Task:
    """Get the current status of a task.

    Args:
        agent_name: The name of the agent that owns the task.
        task_id: The task ID to query.

    Returns:
        Task object with current status, history, and artifacts.

    Raises:
        AuthorizationError: If access to this task is denied by policy.
    """

Example

from agentvisor import a2a

task = a2a.get_task("research-agent", "task-123")
print(f"Status: {task.status.state}")

if task.status.state == "completed":
    for artifact in task.artifacts:
        print(f"Artifact: {artifact.name}")
        for part in artifact.parts:
            if part.text:
                print(part.text)

cancel_task()

def cancel_task(agent_name: str, task_id: str) -> None:
    """Request cancellation of a task.

    Args:
        agent_name: The name of the agent that owns the task.
        task_id: The task ID to cancel.

    Raises:
        AuthorizationError: If this action is denied by policy.
    """

Example

from agentvisor import a2a

# Request cancellation
a2a.cancel_task("research-agent", "task-123")

# Check if cancellation was processed
task = a2a.get_task("research-agent", "task-123")
print(f"Status: {task.status.state}")  # May be "canceled" or still "working"

stream_task()

def stream_task(agent_name: str, task_id: str) -> TaskUpdateIterator:
    """Subscribe to real-time task updates.

    Args:
        agent_name: The name of the agent that owns the task.
        task_id: The task ID to stream updates for.

    Returns:
        TaskUpdateIterator that yields TaskUpdate objects.
    """

Example

from agentvisor import a2a

# Stream updates for a task
for update in a2a.stream_task("research-agent", "task-123"):
    print(f"Event: {update.event_type}")
    print(f"State: {update.task.status.state}")
    
    # Check for terminal states
    if update.task.status.state in ("completed", "failed", "canceled"):
        break

# Or manually control the iterator
stream = a2a.stream_task("research-agent", "task-123")
try:
    update = next(stream)
    print(f"Got update: {update.event_type}")
finally:
    stream.close()  # Always close when done early

text_message()

def text_message(text: str, role: str = "user") -> Message:
    """Create a simple text message.

    Args:
        text: The text content of the message.
        role: The role of the sender ("user" or "agent"). Defaults to "user".

    Returns:
        A Message object with a single text Part.
    """

Example

from agentvisor import a2a

# Create a user message
msg = a2a.text_message("What is the weather today?")

# Send to an agent
task = a2a.send_task("weather-agent", msg)

Task States

Tasks transition through the following states:

State	Description
working	Task is in progress
input_required	Task needs additional input from the caller
auth_required	Task requires authentication
completed	Task finished successfully
failed	Task failed with an error
canceled	Task was canceled
rejected	Task was rejected by the agent

Error Handling

Authorization Errors

from agentvisor import a2a

try:
    card = a2a.get_agent_card("restricted-agent")
except a2a.AuthorizationError as e:
    print(f"Access denied: {e.reason}")

Task Failures

from agentvisor import a2a

task = a2a.send_task("agent", a2a.text_message("do something"))

# Poll until terminal state
while task.status.state == "working":
    task = a2a.get_task("agent", task.id)

if task.status.state == "failed":
    # Get error message from status
    if task.status.message:
        for part in task.status.message.parts:
            if part.text:
                print(f"Error: {part.text}")

Complete Example

from agentvisor import a2a

def process_document(document: str) -> str:
    """Send a document to the research agent for summarization."""
    
    # List available agents
    agents = a2a.list_agents()
    print(f"Available agents: {[a.name for a in agents]}")
    
    # Get agent capabilities
    card = a2a.get_agent_card("research-agent")
    print(f"Capabilities: streaming={card.capabilities.streaming}")
    
    # Create and send task
    message = a2a.text_message(f"Summarize this document:\n\n{document}")
    task = a2a.send_task("research-agent", message)
    print(f"Task started: {task.id}")
    
    # Stream updates
    for update in a2a.stream_task("research-agent", task.id):
        print(f"State: {update.task.status.state}")
        
        if update.task.status.state == "completed":
            # Extract result from artifacts
            for artifact in update.task.artifacts:
                for part in artifact.parts:
                    if part.text:
                        return part.text
            
            # Or from conversation history
            if update.task.history:
                last_msg = update.task.history[-1]
                for part in last_msg.parts:
                    if part.text:
                        return part.text
        
        elif update.task.status.state in ("failed", "rejected"):
            raise RuntimeError(f"Task failed: {update.task.status.state}")
        
        elif update.task.status.state == "canceled":
            raise RuntimeError("Task was canceled")
    
    return ""

Environment

The A2A module requires these environment variables (set automatically by AgentVisor):

Variable	Description
AGENTVISOR_AGENT_SOCKET	Path to agent Unix socket
AGENTVISOR_THREAD_ID	Current thread ID

Related Documentation

• A2A Gateway Guide - Architecture, configuration, and usage
• A2A Transport Guide - Exposing agents via A2A protocol
• Configuration Reference - A2A gateway configuration