A2A Gateway

The A2A Gateway enables agents running in AgentVisor to communicate with external A2A (Agent-to-Agent) compatible agents. This allows your agents to delegate tasks, query capabilities, and collaborate with other AI agents.

Overview

The A2A Gateway implements the Google Agent-to-Agent Protocol, a JSON-RPC 2.0 based protocol for agent-to-agent communication. AgentVisor acts as a gateway, managing connections, credentials, and policy enforcement on behalf of your agents.

┌──────────────────────────────────────────────────────────────────────┐
│  External A2A Agents                                                 │
│                                                                      │
│  ┌──────────────────┐    ┌──────────────────┐    ┌────────────────┐ │
│  │  Research Agent  │    │  Code Agent      │    │  Data Agent    │ │
│  │  /.well-known/   │    │  /.well-known/   │    │  /.well-known/ │ │
│  │ agent-card.json  │    │ agent-card.json  │    │ agent-card.json│ │
│  └────────▲─────────┘    └────────▲─────────┘    └───────▲────────┘ │
│           │                       │                      │          │
│           │        JSON-RPC 2.0 / SSE                    │          │
│           │                       │                      │          │
├───────────┼───────────────────────┼──────────────────────┼──────────┤
│  Host Runtime (A2A Gateway)       │                      │          │
│           │                       │                      │          │
│  ┌────────┴───────────────────────┴──────────────────────┴────────┐ │
│  │  A2A Manager                                                    │ │
│  │                                                                 │ │
│  │  • Connection pooling (LRU cache)                               │ │
│  │  • Credential resolution (per-principal or static)              │ │
│  │  • Retry with exponential backoff                               │ │
│  │  • Policy enforcement via MPE                                   │ │
│  └─────────────────────────────────▲───────────────────────────────┘ │
│                                    │                                 │
│                            Policy Engine                             │
│                                    │                                 │
├────────────────────────────────────┼─────────────────────────────────┤
│  Guest Sandbox (network=none)      │                                 │
│                                    │                                 │
│  ┌─────────────────────────────────┴────────────────────────────┐   │
│  │  Agent Code                                                   │   │
│  │                                                               │   │
│  │  from agentvisor import a2a                                   │   │
│  │  from agentvisor.langchain.a2a import get_a2a_tools           │   │
│  │                                                               │   │
│  │  # Agent calls SDK → gRPC → Gateway → External Agent          │   │
│  └───────────────────────────────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────────────────┘

Configuration

Basic Example

Configure external A2A agents in your agentvisor.yaml:

a2a_gateway:
  agents:
    - name: research-agent
      url: "https://research.example.com"
      credentials:
        type: bearer_token
        source: env
        env_var: RESEARCH_AGENT_API_KEY

Credential Types

The A2A Gateway supports the same credential types as the MCP Gateway:

Bearer Token

Static bearer token from an environment variable:

a2a_gateway:
  agents:
    - name: research-agent
      url: "https://research.example.com"
      credentials:
        type: bearer_token
        source: env
        env_var: RESEARCH_AGENT_TOKEN

API Key

Custom header with API key:

a2a_gateway:
  agents:
    - name: partner-agent
      url: "https://partner.example.com"
      credentials:
        type: api_key
        header_name: X-API-Key
        source: env
        env_var: PARTNER_API_KEY

Principal Passthrough

Forward the caller's JWT to the external agent:

a2a_gateway:
  agents:
    - name: internal-agent
      url: "https://internal.example.com"
      credentials:
        type: principal_passthrough

Token Exchange

RFC 8693 token exchange for scoped tokens:

a2a_gateway:
  agents:
    - name: federated-agent
      url: "https://partner.example.com"
      credentials:
        type: token_exchange
        exchange_url: "https://auth.example.com/oauth/token"
        audience: "partner-agent"
        scope: "a2a:invoke"

With confidential client authentication:

a2a_gateway:
  agents:
    - name: secure-agent
      url: "https://secure.example.com"
      credentials:
        type: token_exchange
        exchange_url: "https://auth.example.com/oauth/token"
        audience: "secure-agent"
        client_id: "agentvisor-client"
        client_secret_env_var: "SECURE_AGENT_CLIENT_SECRET"

Timeout and Retry Configuration

Per-agent timeout and retry settings:

a2a_gateway:
  agents:
    - name: slow-agent
      url: "https://slow.example.com"
      timeout: 120s  # Override default 60s timeout
      retry:
        max_attempts: 5
        initial_interval: 2s
        max_interval: 60s
  
  # Default settings applied to all agents unless overridden
  defaults:
    timeout: 60s
    retry:
      max_attempts: 3
      initial_interval: 1s
      max_interval: 30s

Connection Pool Configuration

For agents using principal-bound credentials (principal_passthrough or token_exchange), connections are managed in an LRU pool:

a2a_gateway:
  pool:
    max_size: 100       # Maximum connections (default: 100)
    idle_timeout: 5m    # Close idle connections (default: 5m)

Usage

Python SDK

The agentvisor.a2a module provides a simple API for communicating with external agents:

from agentvisor import a2a

# List available agents
agents = a2a.list_agents()
for agent_info in agents:
    print(f"{agent_info.name}: {agent_info.description}")

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

# Send a task
message = a2a.text_message("Summarize recent AI safety papers")
task = a2a.send_task("research-agent", message)
print(f"Task ID: {task.id}")
print(f"Status: {task.status.state}")

# Poll for completion
while task.status.state == "working":
    import time
    time.sleep(1)
    task = a2a.get_task("research-agent", task.id)

# Get results
if task.status.state == "completed":
    for artifact in task.artifacts:
        for part in artifact.parts:
            if part.text:
                print(part.text)

Streaming Updates

For real-time task updates, use the streaming API:

from agentvisor import a2a

# Send task
message = a2a.text_message("Analyze this dataset")
task = a2a.send_task("research-agent", message)

# Stream updates
for update in a2a.stream_task("research-agent", task.id):
    print(f"Event: {update.event_type}")
    print(f"State: {update.task.status.state}")
    
    # Check for artifacts
    if update.task.artifacts:
        latest = update.task.artifacts[-1]
        print(f"Artifact: {latest.name}")
    
    # Stop on terminal state
    if update.task.status.state in ("completed", "failed", "canceled"):
        break

Canceling Tasks

from agentvisor import a2a

task = a2a.send_task("research-agent", a2a.text_message("Long running task"))

# Cancel if needed
a2a.cancel_task("research-agent", task.id)

# Verify cancellation
task = a2a.get_task("research-agent", task.id)
print(task.status.state)  # "canceled" or still "working"

Context Management

Use context_id to group related tasks into a conversation:

from agentvisor import a2a

# Start a conversation
task1 = a2a.send_task(
    "research-agent",
    a2a.text_message("Find papers on transformer architectures"),
    context_id="research-session-123"
)

# Continue the conversation
task2 = a2a.send_task(
    "research-agent",
    a2a.text_message("Now focus on attention mechanisms"),
    context_id="research-session-123"  # Same context
)

LangChain Integration

Use A2A agents as LangChain tools:

from agentvisor.langchain.a2a import get_a2a_tools
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

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

# Or get tools for a specific agent's skills
tools = get_a2a_tools("research-agent")

# Create a LangGraph agent
model = ChatAnthropic(model="claude-sonnet-4-20250514")
graph = create_react_agent(model, tools)

# Run the agent
result = graph.invoke({
    "messages": [{"role": "user", "content": "Research recent AI papers"}]
})

Tool Naming

When using get_a2a_tools():

Without agent name: Creates one tool per agent, named a2a_{agent_name}
With agent name: Creates one tool per skill, named {agent_name}_{skill_id}

# Generic tools (one per agent)
tools = get_a2a_tools()
# Tool names: a2a_research_agent, a2a_code_agent

# Skill-specific tools
tools = get_a2a_tools("research-agent")
# Tool names: research_agent_search, research_agent_summarize

Error Handling

Authorization Errors

When a policy denies access:

from agentvisor import a2a
from agentvisor.a2a import AuthorizationError

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

Task Failures

Check task status for failures:

from agentvisor import a2a

task = a2a.get_task("research-agent", "task-123")

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

elif task.status.state == "rejected":
    print("Task was rejected by the agent")

Policy Configuration

The A2A Gateway integrates with AgentVisor's policy engine (MPE). Configure policies to control which agents can access which external agents.

Two-Tier Policy Model

A2A Gateway policy uses a two-tier model that separates coarse gateway-level access from fine-grained per-agent visibility in list responses.

Tier 1 — Gateway-scope check (coarse gate): Applied once before the list or task operation proceeds.

Operation	MRN Scope	Description
`agentvisor:a2a:agent:list`	`mrn:agentvisor:a2a`	Gate entire agent list call
`agentvisor:a2a:agent:describe`	`mrn:agentvisor:a2a:<agent>`	Get agent card (also used as per-item check)
`agentvisor:a2a:task:send`	`mrn:agentvisor:a2a:<agent>`	Send a task to an agent
`agentvisor:a2a:task:get`	`mrn:agentvisor:a2a:<agent>/task/<id>`	Get task status
`agentvisor:a2a:task:cancel`	`mrn:agentvisor:a2a:<agent>/task/<id>`	Cancel a task
`agentvisor:a2a:task:stream`	`mrn:agentvisor:a2a:<agent>/task/<id>`	Stream task updates

Tier 2 — Per-agent check in list responses: When A2AListAgents returns, each agent is individually checked with agentvisor:a2a:agent:describe. Agents the principal cannot access are silently dropped from the response.

Agent list returned from gateway:
  [research-agent, code-agent, admin-agent]
                │
  Per-agent policy check for each:
  research-agent → ALLOWED → included
  code-agent     → ALLOWED → included
  admin-agent    → DENIED  → dropped
                │
                ▼
  Agent SDK receives: [research-agent, code-agent]

agentvisor:a2a:agent:describe gates both A2AGetAgentCard and per-agent visibility in A2AListAgents. A single policy rule covers both discovery and card access.

Onboarding example policy

AgentVisor ships with no default policy. A quick onboarding policy that grants all A2A operations to any authenticated principal via the mrn:agentvisor:a2a:.* wildcard is a common starting point, but is not a production default. Before production use, replace such a wildcard with rules scoped to the specific agents and operations each principal needs, as in the example below.

MRN Formats

Resource	MRN Format	Example
Gateway root (list)	`mrn:agentvisor:a2a`	`mrn:agentvisor:a2a`
Agent	`mrn:agentvisor:a2a:<agent_name>`	`mrn:agentvisor:a2a:research-agent`
Task	`mrn:agentvisor:a2a:<agent_name>/task/<task_id>`	`mrn:agentvisor:a2a:research-agent/task/*`

Example Policy

resources:
  # Allow listing agents and accessing research-agent
  - name: a2a-list
    selector:
      - "mrn:agentvisor:a2a"
    group: "mrn:agentvisor:resourcegroup:allowed"

  - name: research-agent
    selector:
      - "mrn:agentvisor:a2a:research-agent"
      - "mrn:agentvisor:a2a:research-agent/task/.*"
    group: "mrn:agentvisor:resourcegroup:allowed"

  # Block access to admin agent (hidden from list and direct card access)
  - name: restricted-agents
    selector:
      - "mrn:agentvisor:a2a:admin-agent"
    group: "mrn:agentvisor:resourcegroup:denied"

Task States

A2A tasks can be in the following states:

State	Description
`working`	Task is being processed
`input_required`	Agent needs additional input
`auth_required`	Agent requires authentication
`completed`	Task finished successfully
`failed`	Task failed with an error
`canceled`	Task was canceled
`rejected`	Task was rejected by the agent

Terminal states: completed, failed, canceled, rejected

Troubleshooting

Connection Issues

Symptom: Timeout errors, or an unreachable agent, when an agent calls an external A2A agent through the gateway.

A static agent's "connect" at startup only resolves credentials and builds a client — it never dials the agent or fetches its card. A green connected to A2A agent startup log says nothing about whether the agent is actually reachable; the first real network round trip happens on the agent's first live call, or in an explicit probe.

To diagnose:

Run a staged connectivity check against the real agent, using the same config file the runtime loads:
```
agentvisor a2a probe research-agent --config agentvisor.yaml
```
This runs connect → agent-card → tasks/list and reports the first stage that fails (an agent with no tasks/list support is still reported reachable).
Read the failed stage's error classification code: connection_refused, dns_error, tls_error, auth_error, timeout, credential_error, or the connection_error catch-all. See Troubleshooting: Interpreting GET /ready for what each code means and how it's derived.
Check GET /ready for the live picture from the running gateway: an agent that failed to connect at startup, or that degrades at runtime, appears under a2a.failed_endpoints[] with the same classification code, a phase (startup vs runtime), and a since timestamp.
For the verbatim host-side error — hostnames, ports, and the underlying Go error, which the probe output and GET /ready deliberately omit — enable debug logging for the A2A components:
```
AGENTVISOR_LOG_LEVEL=info,a2a=debug,a2a.client=debug,a2a.pool=debug agentvisor serve ./my-agent
```

Also verify the agent URL includes a scheme (https://), and increase timeout in configuration for a genuinely slow agent:

a2a_gateway:
  agents:
    - name: slow-agent
      timeout: 120s

See Troubleshooting Guide: A2A Gateway Issues for connection pool exhaustion, idle timeout, credential resolution failures, and upstream JSON-RPC error codes.

Authentication Failures

Symptom: 401 Unauthorized or 403 Forbidden errors

Solutions:

Verify credentials are set correctly in environment variables
Check token expiration for token_exchange credentials
Review policy configuration for authorization errors

See Troubleshooting Guide: A2A Gateway Issues for the resolver-specific checklist (bearer_token, token_exchange).

Environment Variables

Variable	Default	Description
`AGENTVISOR_A2A_GATEWAY_POOL_MAX_SIZE`	`100`	Maximum connections in the pool
`AGENTVISOR_A2A_GATEWAY_POOL_IDLE_TIMEOUT`	`5m`	Close idle connections after this duration
`AGENTVISOR_A2A_GATEWAY_DEFAULTS_TIMEOUT`	`60s`	Default request timeout
`AGENTVISOR_A2A_GATEWAY_DEFAULTS_RETRY_MAX_ATTEMPTS`	`3`	Maximum retry attempts
`AGENTVISOR_A2A_GATEWAY_DEFAULTS_RETRY_INITIAL_INTERVAL`	`1s`	Initial backoff interval
`AGENTVISOR_A2A_GATEWAY_DEFAULTS_RETRY_MAX_INTERVAL`	`30s`	Maximum backoff interval

Gateways Overview - Understanding gateway architecture
MCP Gateway - Access external MCP servers
A2A Transport - Exposing agents to external A2A clients
Credential Brokering - How credentials are protected
Policy Enforcement - How policies control access
Troubleshooting Guide - Pool exhaustion, idle timeout, credential resolution, and GET /ready
A2A SDK Reference - Complete SDK API documentation
Configuration Reference - Full configuration options

Overview​

Configuration​

Basic Example​

Credential Types​

Bearer Token​

API Key​

Principal Passthrough​

Token Exchange​

Timeout and Retry Configuration​

Connection Pool Configuration​

Usage​

Python SDK​

Streaming Updates​

Canceling Tasks​

Context Management​

LangChain Integration​

Tool Naming​

Error Handling​

Authorization Errors​

Task Failures​

Policy Configuration​

Two-Tier Policy Model​

MRN Formats​

Example Policy​

Task States​

Troubleshooting​

Connection Issues​

Authentication Failures​

Environment Variables​

Related Documentation​