AWS Builder Center

Builder Center

Agent Collaboration: Strands Agents, MCP, and the Agent2Agent Protocol

Agent Collaboration: Strands Agents, MCP, and the Agent2Agent Protocol

This blog post provides an overview of how Strands Agents, an open-source SDK for building and running AI agents, can work with Agent2Agent (A2A) protocol for multi-agent applications.

Jagdeep Singh Soni

The world of AI agents is rapidly evolving, promising increased productivity and automation across various tasks. However, to truly maximize the benefits of agentic AI, these agents need to be able to collaborate effectively, even when built by different teams or using different technologies. This blog post explores how  Strands Agents , a powerful open source SDK, fits into this vision, leveraging  Model Context Protocol (MCP)  tools today and looking towards seamless interoperability through the  Agent2Agent (A2A) protocol  in the future.

Introducing Strands Agents

Strands Agents is an open source SDK designed to simplify the process of building and running AI agents. It takes a model-driven approach, empowering state-of-the-art models to handle tasks like planning, chaining thoughts, calling tools, and reflecting, often with just a prompt and a list of tools. This approach leverages the advanced reasoning and tool use capabilities of modern Large Language Models (LLMs). Strands supports various models from providers like Amazon Bedrock, Anthropic, Meta, and others, and can run in diverse environments. At its core, a Strands agent is defined by a model, tools, and a prompt. The agent operates in a loop, using the model to dynamically direct its steps and interact with tools to complete the specified task.

Strands Agents and MCP Tools

A key aspect of any effective agent is its ability to interact with the outside world, accessing data and performing actions. This is where tools come in. Strands Agents are designed to easily integrate with tools, including those exposed via the Model Context Protocol (MCP).
MCP is an open protocol that standardizes how applications provide context to LLMs. It acts like a universal connector, allowing AI models to plug into various data sources and tools in a standardized way. MCP follows a client-server architecture, where an application (like a Strands agent, acting as an MCP Client) connects to multiple MCP Servers, each exposing specific capabilities securely accessing local data or remote services.
Strands provides flexible tool support. You can easily turn any Python function into a tool using a decorator. For example, defining a tool in Strands looks like this:
Python
@tool
def get_exchange_rate(
    currency_from: str = "USD",
    currency_to: str = "EUR",
    currency_date: str = "latest",
):
    """Tool documentation"""
    # Tool implementation
More importantly for standardized access, Strands can load tools exposed by an MCP server using the  MCPClient . For example, you can use an  MCPClient  to connect to an MCP server. This integration allows Strands agents to leverage a growing ecosystem of pre-built integrations available via MCP, enhancing their ability to access diverse data and capabilities securely.
Python
from mcp import stdio_client, StdioServerParameters
from strands.tools.mcp import MCPClient

stdio_mcp_client = MCPClient(lambda: stdio_client(
    StdioServerParameters(
        command="uvx", 
        args=["awslabs.aws-documentation-mcp-server@latest"]
    )
))

with stdio_mcp_client:
    # Get the tools from the MCP server
    tools = stdio_mcp_client.list_tools_sync()

Introducing the Agent2Agent (A2A) Protocol

While MCP focuses on providing tools and context to an agent, another protocol addresses the challenge of agents needing to collaborate with each other. This is the Agent2Agent (A2A) protocol.
A2A is an open standard designed to enable seamless communication and collaboration between AI agents. Its primary goal is to enable AI agents, even if they were built by different vendors or in different frameworks, to communicate with each other, securely exchange information, and coordinate actions. A2A aims to break down silos and allow agents to collaborate dynamically in a multi-agent ecosystem. This interoperability is seen as crucial for realizing the full potential of collaborative AI agents and automating complex enterprise workflows.

A2A and Strands: Enabling Interoperability

Strands Agents recognizes the importance of agent-to-agent communication. Strands support for the Agent2Agent (A2A) protocol for multi-agent applications is coming soon.
While full, seamless interoperability between Strands Agents and other A2A-compatible agents is coming soon, Strands agents can currently be integrated by implementing the A2A  AgentExecutor  interface. This allows a Strands agent to execute queries received within an A2A context and  return results. 

Implementation Details for A2A Integration

Integrating a Strands agent with the A2A protocol involves implementing the A2A  AgentExecutor  interface. A common pattern is to wrap the core Strands agent logic within a class designed for A2A interaction, such as a  StrandsAgent  wrapper class and a  StrandsAgentExecutor  class.
The  StrandsAgent  class typically initializes and holds the core Strands  Agent  instance, defining how it can be invoked. It can provide methods for both synchronous invocation and streaming interactions.
Python
class StrandsAgent:
    def __init__(self):
        bedrock_model = BedrockModel(
            model_id="us.anthropic.claude-3-5-haiku-20241022-v1:0",
            region_name="us-east-1",
            temperature=0.3,
        )
        self.agent = Agent(
            tools=[get_exchange_rate], # Tools are provided to the Strands Agent
            model=bedrock_model,
            system_prompt="Your system prompt here"
        )

    async def invoke(self, query: str) -> str:
        """Invoke the agent with a query and return the result as a string."""
        result = self.agent(query) # Invoking the underlying Strands agent
        return str(result)

    async def stream(self, query: str) -> AsyncIterable[Dict[str, Any]]:
        """Stream the agent's response using stream_async."""
        async for event in self.agent.stream_async(query):
            # Process streaming events
            # ...
The  StrandsAgentExecutor  class then implements the A2A  AgentExecutor  interface, using the  StrandsAgent  wrapper internally. The  execute  method is a core part of this interface, where the Strands agent is invoked and its results are communicated back via the A2A  EventQueue .
Python
class StrandsAgentExecutor(AgentExecutor): # Implementing the A2A AgentExecutor interface
    def __init__(self):
        self.agent = StrandsAgent() # Using the StrandsAgent wrapper

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        query = "Your query here" # The query can be derived from the A2A context
        result = await self.agent.invoke(query) # Invoking the Strands agent
        event_queue.enqueue_event(new_agent_text_message(result)) # Sending result via A2A EventQueue

The Power of Interoperability: Building Collaborative Agentic Applications

The combination of Strands' agent building capabilities and the interoperability offered by A2A unlocks significant possibilities for building complex, multi-agent systems.
Customers using Strands will be able to:
  • Break Down Silos: Connect their Strands agents with specialized agents built on other frameworks or by different vendors, as long as those agents support A2A.
  • Enable Complex Collaboration: Orchestrate Strands agents to work alongside other A2A-compatible agents to tackle tasks that no single agent could handle alone.
  • Build Richer Agentic Applications: Combine the strengths of various agents – for example, a Strands agent could coordinate tasks with an agent built using another framework like LangGraph. Within Strands itself, developers can already apply advanced techniques for multi-agent systems using built-in "multi-agent tools" like the workflow, graph, and swarm tools.
The vision is a future where agents, regardless of how they were built, can discover each other's capabilities, securely exchange information, and coordinate actions. Strands Agents, with its focus on ease of use and production readiness, is poised to play a key role in this interoperable ecosystem, leveraging MCP for tools and integrating with A2A for agent-to-agent communication.

Getting Started with Strands and A2A

For detail guide on A2A setup, you can check  A2A QuickStart Tutorial  and  samples . To begin integrating Strands Agents with A2A, you can follow these general steps outlined below:
  • Install dependencies: You'll need the  strands  and  a2a  libraries.
Python
pip install strands a2a
  • Create your Strands agent implementation: Define your agent logic, potentially within classes like  StrandsAgent  and  StrandsAgentExecutor .
Python
# agent_with_mcp.py

from collections.abc import AsyncIterable
from typing import Dict, Any, List, Optional
import logging
import json

import httpx
from strands import Agent
from strands.models import BedrockModel
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.utils import new_agent_text_message

# Set up logging
logger = logging.getLogger(__name__)
if not logger.handlers:
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)

class StrandsAgent:
    """Strands Agent with streaming capability and MCP tool support for Airbnb searches."""

    def __init__(self, mcp_tools: Optional[List[Any]] = None):
        """
        Initialize the Strands Agent with optional MCP tools.
        
        Args:
            mcp_tools: Optional list of MCP tools to be used by the agent.
        """
        logger.info("Initializing StrandsAgent with Bedrock model")
        
        # Initialize Bedrock model
        bedrock_model = BedrockModel(
            model_id="us.anthropic.claude-3-5-haiku-20241022-v1:0",
            region_name="us-east-1",
            temperature=0.3,
        )
        
        # Use MCP tools if provided
        tools = []
        if mcp_tools:
            logger.info(f"Adding {len(mcp_tools)} MCP tools to the agent")
            tools.extend(mcp_tools)
        else:
            logger.info("No MCP tools provided, agent functionality will be limited")
        
        # Create system prompt based on available tools
        system_prompt = (
            "You are a specialized assistant for Airbnb accommodations. Your primary function is to utilize "
            "the provided tools to search for Airbnb listings and answer related questions. You must rely "
            "exclusively on these tools for information; do not invent listings or prices. Ensure that your "
            "Markdown-formatted response includes all relevant tool output, with particular emphasis on "
            "providing direct links to listings. When you have a final answer for the user and have used "
            "tools sufficiently, provide that answer directly without calling more tools."
        )
        
        # Initialize the Strands Agent with tools and model
        self.agent = Agent(
            tools=tools,
            model=bedrock_model,
            system_prompt=system_prompt,
        )
        
        logger.info("StrandsAgent initialized successfully")

    async def invoke(self, query: str) -> str:
        """
        Invoke the agent with a query and return the result.
        
        Args:
            query: The user's query string.
            
        Returns:
            The agent's response as a string.
        """
        logger.info(f"Invoking agent with query: {query}")
        try:
            result = self.agent(query)
            logger.info("Agent invocation completed successfully")
            return result
        except Exception as e:
            logger.error(f"Error invoking agent: {e}", exc_info=True)
            raise

    async def stream(self, query: str) -> AsyncIterable[Dict[str, Any]]:
        """
        Stream the agent's response using stream_async.
        
        Args:
            query: The user's query string.
            
        Yields:
            Dictionary containing streaming response chunks.
        """
        logger.info(f"Streaming agent response for query: {query}")
        try:
            async for event in self.agent.stream_async(query):
                # Handle text content
                if isinstance(event, dict) and "data" in event and isinstance(event["data"], str):
                    yield {
                        "is_task_complete": False,
                        "require_user_input": False,
                        "content": event["data"],
                    }
                # Handle content block delta events
                elif (
                    isinstance(event, dict)
                    and "event" in event
                    and "contentBlockDelta" in event["event"]
                    and "delta" in event["event"]["contentBlockDelta"]
                    and "text" in event["event"]["contentBlockDelta"]["delta"]
                ):
                    text = event["event"]["contentBlockDelta"]["delta"]["text"]
                    if text:
                        yield {
                            "is_task_complete": False,
                            "require_user_input": False,
                            "content": text,
                        }
                # Handle tool calls
                elif (
                    isinstance(event, dict)
                    and "data" in event
                    and isinstance(event["data"], dict)
                    and "tool_calls" in event["data"]
                ):
                    tool_name = event["data"]["tool_calls"][0].get("name", "unknown tool")
                    yield {
                        "is_task_complete": False,
                        "require_user_input": False,
                        "content": f"Using {tool_name}...",
                    }
                # Handle tool results
                elif (
                    isinstance(event, dict)
                    and "data" in event
                    and isinstance(event["data"], dict)
                    and "tool_results" in event["data"]
                ):
                    yield {
                        "is_task_complete": False,
                        "require_user_input": False,
                        "content": "Processing results...",
                    }

            # Final chunk indicating completion
            logger.info("Agent streaming completed")
            yield {"is_task_complete": True, "require_user_input": False, "content": ""}
        except Exception as e:
            logger.error(f"Error during agent streaming: {e}", exc_info=True)
            yield {
                "is_task_complete": True,
                "require_user_input": False,
                "content": f"An error occurred: {str(e)}",
            }

class StrandsAgentExecutor(AgentExecutor):
    """Agent Executor for Strands Agent with MCP tool support."""

    def __init__(self, mcp_tools: Optional[List[Any]] = None):
        """
        Initialize the StrandsAgentExecutor with optional MCP tools.
        
        Args:
            mcp_tools: Optional list of MCP tools to be used by the agent.
        """
        logger.info("Initializing StrandsAgentExecutor")
        self.agent = StrandsAgent(mcp_tools=mcp_tools)

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        """
        Execute the agent using the invoke method.
        
        Args:
            context: The request context.
            event_queue: The event queue for sending events.
        """
        logger.info(f"Executing agent with context: {context}")
        
        # Debug the context structure
        logger.info(f"Context attributes: {dir(context)}")
        logger.info(f"Context dict: {vars(context)}")
        
        # Extract query from context - try different possible structures
        query = None
        
        # Try to get content from message parts if available
        if hasattr(context, 'message') and hasattr(context.message, 'parts'):
            for part in context.message.parts:
                if hasattr(part, 'kind') and part.kind == 'text' and hasattr(part, 'text'):
                    query = part.text
                    break
        
        # If not found, try direct content attribute
        if not query and hasattr(context, 'content'):
            query = context.content
            
        # If still not found, check if there's a request object
        if not query and hasattr(context, 'request') and hasattr(context.request, 'content'):
            query = context.request.content
            
        # If still not found, use default query
        if not query:
            query = "Find me a place to stay in New York for next weekend"
            
        logger.info(f"Using query: {query}")
        
        try:
            # Use invoke method to get response
            result = await self.agent.invoke(query)
            
            # Extract text content from result
            if hasattr(result, "message") and isinstance(result.message, dict) and "content" in result.message:
                content = result.message["content"]
                if isinstance(content, list) and len(content) > 0 and "text" in content[0]:
                    response_text = content[0]["text"]
                    logger.info(f"Agent response: {response_text[:100]}...")
                    event_queue.enqueue_event(new_agent_text_message(response_text))
                else:
                    logger.warning("Unexpected result format, using string representation")
                    event_queue.enqueue_event(new_agent_text_message(str(result)))
            else:
                logger.warning("Unexpected result format, using string representation")
                event_queue.enqueue_event(new_agent_text_message(str(result)))
        except Exception as e:
            logger.error(f"Error executing agent: {e}", exc_info=True)
            event_queue.enqueue_event(
                new_agent_text_message(f"An error occurred: {str(e)}")
            )

    async def execute_stream(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        """
        Execute the agent using the stream method.
        
        Args:
            context: The request context.
            event_queue: The event queue for sending events.
        """
        logger.info(f"Executing agent with streaming, context: {context}")
        
        # Debug the context structure
        logger.info(f"Context attributes: {dir(context)}")
        
        # Extract query using the same approach as in execute
        query = None
        
        # Try to get content from message parts if available
        if hasattr(context, 'message') and hasattr(context.message, 'parts'):
            for part in context.message.parts:
                if hasattr(part, 'kind') and part.kind == 'text' and hasattr(part, 'text'):
                    query = part.text
                    break
        
        # If not found, try direct content attribute
        if not query and hasattr(context, 'content'):
            query = context.content
            
        # If still not found, check if there's a request object
        if not query and hasattr(context, 'request') and hasattr(context.request, 'content'):
            query = context.request.content
            
        # If still not found, use default query
        if not query:
            query = "Find me a place to stay in New York for next weekend"
            
        logger.info(f"Using query for streaming: {query}")
        
        try:
            # Stream response chunks
            async for chunk in self.agent.stream(query):
                if "content" in chunk and chunk["content"]:
                    event_queue.enqueue_event(new_agent_text_message(chunk["content"]))
        except Exception as e:
            logger.error(f"Error during streaming execution: {e}", exc_info=True)
            event_queue.enqueue_event(
                new_agent_text_message(f"An error occurred during streaming: {str(e)}")
            )

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        """
        Cancel the agent execution.
        
        Args:
            context: The request context.
            event_queue: The event queue for sending events.
        """
        logger.warning("Cancel operation requested but not supported")
        event_queue.enqueue_event(
            new_agent_text_message("Cancel operation is not supported")
        )

  • Configure Strands Agent with A2A server
Python
# __main__.py

import os
import sys
from typing import Dict, Any, List
import asyncio
from contextlib import asynccontextmanager
import click
import uvicorn
from agent_with_mcp import StrandsAgentExecutor
from dotenv import load_dotenv

from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)
from a2a.server.tasks import InMemoryTaskStore

from mcp import stdio_client, StdioServerParameters
from strands.tools.mcp import MCPClient

load_dotenv(override=True)

app_context: Dict[str, Any] = {}

@asynccontextmanager
async def app_lifespan(context: Dict[str, Any]):
    """Manages the lifecycle of shared resources like the MCP client and tools."""
    print("Lifespan: Initializing MCP client and tools...")

    # This variable will hold the MCPClient instance
    mcp_client_instance: MCPClient | None = None

    try:
        # Initialize MCPClient with Airbnb MCP server
        mcp_client_instance = MCPClient(
            lambda: stdio_client(
                StdioServerParameters(
                    command="npx",
                    args=["-y", "@openbnb/mcp-server-airbnb", "--ignore-robots-txt"],
                )
            )
        )

        # Start the MCP client
        mcp_client_instance.__enter__()

        # Get tools from the MCP client
        mcp_tools = mcp_client_instance.list_tools_sync()
        context["mcp_tools"] = mcp_tools

        tool_count = len(mcp_tools) if mcp_tools else 0
        print(f"Lifespan: MCP Tools preloaded successfully ({tool_count} tools found).")
        yield  # Application runs here
    except Exception as e:
        print(f"Lifespan: Error during initialization: {e}", file=sys.stderr)
        raise
    finally:
        print("Lifespan: Shutting down MCP client...")
        if mcp_client_instance:
            try:
                # Clean up the MCP client
                mcp_client_instance.__exit__(None, None, None)
                print("Lifespan: MCP Client resources released.")
            except Exception as e:
                print(
                    f"Lifespan: Error during MCP client cleanup: {e}",
                    file=sys.stderr,
                )

        # Clear the application context
        print("Lifespan: Clearing application context.")
        context.clear()

@click.command()
@click.option(
    "--host", "host", default="localhost", help="Hostname to bind the server to."
)
@click.option(
    "--port", "port", default=10002, type=int, help="Port to bind the server to."
)
@click.option("--log-level", "log_level", default="info", help="Uvicorn log level.")
def cli_main(host: str, port: int, log_level: str):
    """Command Line Interface to start the Strands Agent server with MCP tools."""

    async def run_server_async():
        async with app_lifespan(app_context):
            if not app_context.get("mcp_tools"):
                print(
                    "Warning: MCP tools were not loaded. Agent will use default tools only.",
                    file=sys.stderr,
                )

            # Initialize StrandsAgentExecutor with preloaded MCP tools
            strands_agent_executor = StrandsAgentExecutor(
                mcp_tools=app_context.get("mcp_tools", [])
            )

            request_handler = DefaultRequestHandler(
                agent_executor=strands_agent_executor,
                task_store=InMemoryTaskStore(),
            )

            # Create the A2AServer instance
            a2a_server = A2AStarletteApplication(
                agent_card=get_agent_card(host, port), http_handler=request_handler
            )

            # Get the ASGI app from the A2AServer instance
            asgi_app = a2a_server.build()

            config = uvicorn.Config(
                app=asgi_app,
                host=host,
                port=port,
                log_level=log_level.lower(),
                lifespan="auto",
            )

            uvicorn_server = uvicorn.Server(config)

            print(
                f"Starting Uvicorn server at http://{host}:{port} with log-level {log_level}..."
            )
            try:
                await uvicorn_server.serve()
            except KeyboardInterrupt:
                print("Server shutdown requested (KeyboardInterrupt).")
            finally:
                print("Uvicorn server has stopped.")

    try:
        asyncio.run(run_server_async())
    except RuntimeError as e:
        if "cannot be called from a running event loop" in str(e):
            print(
                "Critical Error: Attempted to nest asyncio.run(). This should have been prevented.",
                file=sys.stderr,
            )
        else:
            print(f"RuntimeError in cli_main: {e}", file=sys.stderr)
        sys.exit(1)
    except Exception as e:
        print(f"An unexpected error occurred in cli_main: {e}", file=sys.stderr)
        sys.exit(1)

def get_agent_card(host: str, port: int):
    """Returns the Agent Card for the Airbnb Agent."""
    capabilities = AgentCapabilities(streaming=True, pushNotifications=True)
    skill = AgentSkill(
        id="airbnb_search",
        name="Search airbnb accommodation",
        description="Helps with accommodation search using airbnb",
        tags=["airbnb accommodation"],
        examples=[
            "Please find a room in LA, CA, April 15, 2025, checkout date is april 18, 2 adults",
            "Find me a place to stay in New York for next weekend",
            "What are the best Airbnb options in Miami for a family of 4?",
        ],
    )
    return AgentCard(
        name="Airbnb Agent",
        description="Helps with searching accommodation",
        url=f"http://{host}:{port}/",
        version="1.0.0",
        defaultInputModes=["text"],
        defaultOutputModes=["text"],
        capabilities=capabilities,
        skills=[skill],
    )

if __name__ == "__main__":
    cli_main()

  • Start the A2A server: This makes your Strands-based A2A agent available.
Python
uv run .
  • Test with an A2A client: Use an A2A client implementation to interact with your agent.
Python
from typing import Any
from uuid import uuid4

import httpx

from a2a.client import A2AClient
from a2a.types import (
    SendMessageResponse,
    GetTaskResponse,
    SendMessageSuccessResponse,
    Task,
    SendMessageRequest,
    MessageSendParams,
    GetTaskRequest,
    TaskQueryParams,
)
import traceback

AGENT_URL = "http://localhost:10002"

def create_send_message_payload(
    text: str, task_id: str | None = None, context_id: str | None = None
) -> dict[str, Any]:
    """Helper function to create the payload for sending a task."""
    payload: dict[str, Any] = {
        'message': {
            'role': 'user',
            'parts': [{'kind': 'text', 'text': text}],
            'messageId': uuid4().hex,
        },
    }

    if task_id:
        payload['message']['taskId'] = task_id

    if context_id:
        payload['message']['contextId'] = context_id
    return payload

def print_json_response(response: Any, description: str) -> None:
    """Helper function to print the JSON representation of a response."""
    print(f"--- {description} ---")
    if hasattr(response, "root"):
        print(f"{response.root.model_dump_json(exclude_none=True)}\n")
    else:
        print(f'{response.model_dump(mode="json", exclude_none=True)}\n')

async def run_single_turn_test(client: A2AClient) -> None:
    """Runs a single-turn non-streaming test."""

    send_payload = create_send_message_payload(
        text='Please find a bedroom in new york city, June 15-20, 2025'
    )
    request = SendMessageRequest(id=str(uuid4()), params=MessageSendParams(**send_payload))

    print('--- Single Turn Request ---')
    # Send Message
    send_response: SendMessageResponse = await client.send_message(request)
    print_json_response(send_response, 'Single Turn Request Response')
    if not isinstance(send_response.root, SendMessageSuccessResponse):
        print('received non-success response. Aborting get task ')
        return

    if not isinstance(send_response.root.result, Task):
        print('received non-task response. Aborting get task ')
        return

    task_id: str = send_response.root.result.id
    print('---Query Task---')
    # query the task
    get_request = GetTaskRequest(id=str(uuid4()), params=TaskQueryParams(id=task_id))
    get_response: GetTaskResponse = await client.get_task(get_request)
    print_json_response(get_response, 'Query Task Response')

    
async def main() -> None:
    """Main function to run the tests."""
    print(f'Connecting to agent at {AGENT_URL}...')
    try:
        async with httpx.AsyncClient(timeout=30) as httpx_client:
            client = await A2AClient.get_client_from_agent_card_url(
                httpx_client, AGENT_URL
            )
            print('Connection successful.')

            await run_single_turn_test(client)

    except Exception as e:
        traceback.print_exc()
        print(f'An error occurred: {e}')
        print('Ensure the agent server is running.')

if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

Python
uv run test_client.py
These steps provide a basic pathway to setting up and testing a Strands agent integrated as an A2A endpoint.

Conclusion

By integrating with the MCP for tool access and embracing the upcoming Agent2Agent protocol for seamless inter-agent communication, Strands empowers developers to build complex, multi-agent systems that can tackle tasks beyond the capabilities of any single agent.
Strands Agents, with its focus on ease of use and production readiness, is poised to play a key role in this emerging interoperable ecosystem. The future envisioned is one where AI agents, much like human teams, can discover each other's capabilities, securely exchange information, and coordinate their actions to solve problems more effectively.
These steps provide a basic pathway to setting up and testing a Strands agent integrated with A2A.

Authors

 Jagdeep Singh Soni  is a Senior Partner Solutions Architect at AWS based in the Netherlands. He uses his passion for generative AI to help customers and partners build generative AI applications using AWS services. Jagdeep has 15 years of experience in innovation, experience engineering, digital transformation, cloud architecture, and ML applications.
 Ajeet Tewari  is a Senior Solutions Architect for Amazon Web Services. He works with enterprise customers to help them navigate their journey to AWS. His specialties include architecting and implementing scalable OLTP systems and leading strategic AWS initiatives.  
# strands# a2a# mcp# genai
Any opinions in this article are those of the individual author and may not reflect the opinions of AWS.
Enjoyed reading this content? Let the author know!

Your likes, comments, shares, and saves help creators reach more builders.

Loading recommendations

Loading article