Agent-to-Agent (A2A) Protocol

The Agent-to-Agent (A2A) protocol lets your Strands agents call agents built on other platforms and frameworks, and lets those agents call yours. Strands supports A2A on both sides: consuming a remote agent as a client, and serving one of your own agents over the protocol.

What is Agent-to-Agent (A2A)?

The Agent-to-Agent protocol is an open standard that defines how AI agents can discover, communicate, and collaborate with each other.

Use Cases

A2A support covers several patterns:

• Multi-Agent Workflows: Chain multiple specialized agents together
• Agent Marketplaces: Discover and use agents from different providers
• Cross-Platform Integration: Connect Strands agents with other A2A-compatible systems
• Distributed AI Systems: Build scalable, distributed agent architectures

Learn more about the A2A protocol:

• A2A Documentation
• A2A GitHub Organization

Complete Examples Available

Check out the Native A2A Support samples for complete, ready-to-run client, server and tool implementations.

Installation

To use A2A functionality with Strands, install the package with the A2A dependencies:

Python

pip install 'strands-agents[a2a]'

This installs the core Strands Harness SDK along with the necessary A2A protocol dependencies.

TypeScript

npm install @strands-agents/sdk @a2a-js/sdk express

@a2a-js/sdk and express are optional peer dependencies of @strands-agents/sdk and must be installed explicitly.

Consuming Remote Agents

You consume a remote A2A agent through the A2AAgent class. It wraps the protocol communication behind a familiar interface: you invoke it just like a regular Strands Agent, without resolving agent cards, configuring HTTP clients, building protocol messages, or parsing responses by hand.

Basic Usage

Python

from strands.agent.a2a_agent import A2AAgent

# Create an A2AAgent pointing to a remote A2A server
a2a_agent = A2AAgent(endpoint="http://localhost:9000")

# Invoke it just like a regular Agent
result = a2a_agent("Show me 10 ^ 6")
print(result.message)
# {'role': 'assistant', 'content': [{'text': '10^6 = 1,000,000'}]}

TypeScript

import { A2AAgent } from '@strands-agents/sdk/a2a'

// Create an A2AAgent pointing to a remote A2A server
const a2aAgent = new A2AAgent({ url: 'http://localhost:9000' })

// Invoke it just like a regular Agent
const result = await a2aAgent.invoke('Show me 10 ^ 6')
console.log(result.lastMessage.content)

The A2AAgent returns an AgentResult, the same type a local Agent returns, so a remote agent drops into code that already handles agent results.

Configuration Options

Python

The A2AAgent constructor accepts these parameters.

Parameter	Type	Default	Description
endpoint	str	Required	Base URL of the remote A2A agent
name	str	None	Agent name (auto-populated from agent card if not provided)
description	str	None	Agent description (auto-populated from agent card if not provided)
timeout	int	300	Timeout for HTTP operations in seconds
client_config	ClientConfig	None	A2A client config for authenticated transport (SigV4, OAuth, bearer tokens); its httpx_client handles both card discovery and message sending

To reach an authenticated endpoint, pass a ClientConfig with a configured httpx_client. The a2a_client_factory parameter is deprecated; use client_config instead.

TypeScript

The A2AAgent constructor accepts a config object with these properties.

Property	Type	Default	Description
url	string	Required	Base URL of the remote A2A agent
agentCardPath	string	/.well-known/agent-card.json	Path to the agent card endpoint
id	string	The url value	Unique identifier for the agent instance
name	string	From agent card	Agent name (auto-populated from agent card if not provided)
description	string	From agent card	Agent description (auto-populated from agent card if not provided)
clientFactory	ClientFactory	None	Custom A2A client factory for authenticating requests (SigV4, bearer token)

The agent card is fetched lazily on the first invoke() or stream() call.

Asynchronous Invocation

Python

For async workflows, use invoke_async:

import asyncio
from strands.agent.a2a_agent import A2AAgent

async def main():
    a2a_agent = A2AAgent(endpoint="http://localhost:9000")
    result = await a2a_agent.invoke_async("Calculate the square root of 144")
    print(result.message)

asyncio.run(main())

TypeScript

In TypeScript, invoke is always async:

import { A2AAgent } from '@strands-agents/sdk/a2a'

const a2aAgent = new A2AAgent({ url: 'http://localhost:9000' })
const result = await a2aAgent.invoke('Calculate the square root of 144')
console.log(result.lastMessage.content)

Streaming Responses

Python

For real-time streaming of responses, use stream_async:

import asyncio
from strands.agent.a2a_agent import A2AAgent

async def main():
    a2a_agent = A2AAgent(endpoint="http://localhost:9000")

    async for event in a2a_agent.stream_async("Explain quantum computing"):
        if "data" in event:
            print(event["data"], end="", flush=True)

asyncio.run(main())

TypeScript

const remoteAgent = new A2AAgent({ url: 'http://localhost:9000' })

// stream() yields A2AStreamUpdateEvent for each protocol event,
// then an AgentResultEvent with the final result
const stream = remoteAgent.stream('Explain quantum computing')
let next = await stream.next()
while (!next.done) {
  console.log(next.value)
  next = await stream.next()
}
// Final result
console.log(next.value)

A2AAgent.stream() uses sendMessageStream from the A2A SDK. It yields A2AStreamUpdateEvent for each protocol event (messages, task status updates, artifact updates) followed by an AgentResultEvent with the final result.

Fetching the Agent Card

Python

You can retrieve the remote agent’s metadata using get_agent_card:

import asyncio
from strands.agent.a2a_agent import A2AAgent

async def main():
    a2a_agent = A2AAgent(endpoint="http://localhost:9000")
    card = await a2a_agent.get_agent_card()
    print(f"Agent: {card.name}")
    print(f"Description: {card.description}")
    print(f"Skills: {card.skills}")

asyncio.run(main())

TypeScript

The agent card is fetched and cached internally on the first invoke() or stream() call. There is no separate public method to retrieve it.

A2AAgent in Multi-Agent Patterns

The A2AAgent class integrates with Strands multi-agent patterns that support it. Currently, you can use remote A2A agents in Graph workflows (Python only) and as tools in an orchestrator agent.

As a Tool

You can wrap an A2AAgent as a tool in an orchestrator agent’s toolkit:

Python

from strands import Agent, tool
from strands.agent.a2a_agent import A2AAgent

calculator_agent = A2AAgent(
    endpoint="http://calculator-service:9000",
    name="calculator"
)

@tool
def calculate(expression: str) -> str:
    """Perform a mathematical calculation."""
    result = calculator_agent(expression)
    return str(result.message["content"][0]["text"])

orchestrator = Agent(
    system_prompt="You are a helpful assistant. Use the calculate tool for math.",
    tools=[calculate]
)

TypeScript

const calculatorAgent = new A2AAgent({
  url: 'http://calculator-service:9000',
})

const calculate = tool({
  name: 'calculate',
  description: 'Perform a mathematical calculation.',
  inputSchema: z.object({
    expression: z.string().describe('The math expression to evaluate'),
  }),
  callback: async (input) => {
    const calcResult = await calculatorAgent.invoke(input.expression)
    return String(calcResult.lastMessage.content[0])
  },
})

const orchestrator = new Agent({
  systemPrompt: 'You are a helpful assistant. Use the calculate tool for math.',
  tools: [calculate],
})

In Graph Workflows

The A2AAgent works as a node in Graph workflows. See Remote Agents with A2AAgent for detailed examples of mixing local and remote agents in graph-based pipelines.

In Swarm Patterns

Not yet supported

A2AAgent is not currently supported in Swarm patterns in either SDK. Swarm coordination relies on tool-based handoffs that require capabilities not yet available in the A2A protocol. Use Graph workflows for multi-agent patterns with remote A2A agents.

Creating an A2A Server

Basic Server Setup

Create a Strands agent and expose it as an A2A server:

Python

import logging
from strands import Agent
from strands.multiagent.a2a import A2AServer
from strands.vended_tools import notebook

logging.basicConfig(level=logging.INFO)

# Build a fresh agent for each A2A context so callers stay isolated
def create_agent(context_id: str) -> Agent:
    return Agent(
        name="Notebook Agent",
        description="A note-taking agent that organizes information in notebooks.",
        tools=[notebook],
        callback_handler=None
    )

# Create A2A server with a per context agent factory (streaming enabled by default)
a2a_server = A2AServer(agent_factory=create_agent)

# Start the server
a2a_server.serve()

TypeScript

import { Agent } from '@strands-agents/sdk'
import { A2AExpressServer } from '@strands-agents/sdk/a2a/express'

// Build a fresh agent for each A2A context so callers stay isolated
const server = new A2AExpressServer({
  agentFactory: (contextId) =>
    new Agent({
      systemPrompt: 'You are a calculator agent that can perform basic arithmetic.',
    }),
  name: 'Calculator Agent',
  description: 'A calculator agent that can perform basic arithmetic operations.',
})

await server.serve()

The server serves the agent card at /.well-known/agent-card.json and handles JSON-RPC requests at the root path. Streaming is supported by default.

Conversation Isolation

The A2A protocol identifies each conversation with a context_id. The server isolates conversation state per context_id so that callers in different contexts never read or influence each other’s history. Two modes control how that isolation works.

The recommended mode is agent_factory - you provide a callable that takes a context_id and returns a dedicated agent for each conversation, reusing it for later requests in that context. This enables different conversations to run concurrently because each owns an independent agent.

The factory is also where you wire per-conversation concerns such as a session_manager to persist that conversation’s history.

Python

from strands import Agent
from strands.multiagent.a2a import A2AServer

def create_agent(context_id: str) -> Agent:
    return Agent(
        name="Calculator Agent",
        description="A calculator agent.",
        callback_handler=None
    )

a2a_server = A2AServer(agent_factory=create_agent)
a2a_server.serve()

TypeScript

import { Agent, SessionManager, FileStorage } from '@strands-agents/sdk'
import { A2AExpressServer } from '@strands-agents/sdk/a2a/express'

// The factory runs once per contextId and returns a dedicated agent, so each conversation
// is isolated. Wire an optional sessionManager here to persist that conversation's history,
// scoped to the contextId.
const storage = new FileStorage('./sessions')

const server = new A2AExpressServer({
  agentFactory: (contextId) =>
    new Agent({
      name: 'Calculator Agent',
      description: 'A calculator agent.',
      sessionManager: new SessionManager({
        sessionId: contextId,
        storage: { snapshot: storage },
      }),
    }),
  name: 'Calculator Agent',
  maxContexts: 1000,
})

await server.serve()

The server retains at most max_contexts contexts at once (default 1000). When that cap is exceeded, the server evicts the least recently used context, and a later request that reuses the evicted context_id starts a fresh conversation. Tune this cap to bound memory in long running servers.

context_id is not an authentication boundary

Contexts are keyed on the client supplied context_id. A caller who knows another caller’s context_id can attach to that conversation. Multi-tenant deployments must enforce authenticated identity at the transport or gateway layer.

Passing a single agent is deprecated

In single-agent mode the server reuses one agent across every context, swapping each context’s saved state on and off the shared instance under a lock, which serializes all requests. A single agent with a configured session_manager is rejected, because the session manager would persist every context into one interleaved session. Prefer agent_factory for any deployment that serves more than one conversation.

Answering Interrupts

Python only

This feature is currently available in the Python SDK only.

When a tool or hook on the served agent raises an interrupt, the task moves to the A2A input_required state and waits. The client answers it, and the task resumes the paused tool exactly where it stopped.

Answering an interrupt is the one flow on this page that needs the raw a2a-sdk client rather than A2AAgent. A2AAgent speaks in text, so it raises ValueError if you pass it interrupt responses, and it drops the DataPart carrying the interrupt ids when it reads the reply. The examples below build A2A messages directly.

Each interrupt has a server-generated id, and an answer is bound to the id of the interrupt that raised it. The server advertises the pending interrupts on the input_required status message as a DataPart, alongside the human-readable TextPart:

{
  "kind": "data",
  "data": {
    "interrupts": [
      {
        "interruptId": "v1:tool_call:tu-1:a71adb48-e65d-55fa-b155-0359c9cd3b66",
        "name": "approve_campaign",
        "reason": {"name": "spring-launch"}
      }
    ]
  }
}

To answer, send a new message on the same taskId containing a DataPart that echoes the interruptId back with the response:

{
  "kind": "data",
  "data": {
    "interruptResponse": {
      "interruptId": "v1:tool_call:tu-1:a71adb48-e65d-55fa-b155-0359c9cd3b66",
      "response": {"approved": true}
    }
  }
}

The response becomes the return value of the interrupt() call that paused the tool. It is any JSON value except null, which the server refuses: a null answer would leave the interrupt unsatisfied and re-raise it. false and 0 are fine. Answer several interrupts in one message by sending one DataPart for each.

Reading the ids off the status message and answering them:

from a2a.types import DataPart, Part

# The task parked in input_required; read the interrupts it is waiting on.
pending = next(
    part.root.data["interrupts"]
    for part in task.status.message.parts
    if isinstance(part.root, DataPart) and "interrupts" in part.root.data
)

# Answer each one on the same taskId.
answers = [
    Part(root=DataPart(data={
        "interruptResponse": {"interruptId": item["interruptId"], "response": {"approved": True}}
    }))
    for item in pending
]

The server rejects an answer it cannot bind, before the agent runs, so a refused answer leaves the interrupt pending and the task still answerable. An answer is rejected when:

• its interruptId does not match an interrupt the task is waiting on
• no interrupt is pending
• the same interruptId is answered twice in one message
• the payload is missing interruptId, or its response is missing or null
• it is sent alongside other content parts

A task with a pending interrupt also rejects an ordinary conversational message. Answer the interrupt, or cancel the task.

A DataPart without an interruptResponse key is unaffected and continues to reach the agent as structured data.

Server Configuration and Deployment

The A2AServer (Python) and A2AExpressServer (TypeScript) expose the full set of constructor options, custom task stores and request-handler components, and path-based mounting for load-balanced deployments. See A2A Server Configuration for the complete reference.

Troubleshooting

If you encounter bugs or need to request features for A2A support:

1. Check the A2A documentation for protocol-specific issues
2. Report Strands-specific issues on GitHub
3. Include relevant error messages and code samples in your reports

Tags

multi-agent8