Forem: cnych

A2A Protocol Explained

cnych — Wed, 16 Jul 2025 08:38:54 +0000

A2A, short for Agent to Agent protocol, is an open-source framework launched by Google to facilitate communication and interoperability among AI agents. By providing a standardized collaboration method for agents, regardless of their underlying frameworks or vendors, this protocol enables AI agents to securely exchange information, coordinate actions, and operate across diverse enterprise platforms and applications. In simple terms, it addresses the question: How can AI agents developed by different teams, using different technologies, and owned by different organizations effectively communicate and collaborate?

Why A2A is Needed

As AI agents become increasingly specialized and powerful, the need for them to collaborate on complex tasks grows. Imagine a user requesting their primary AI agent to plan an international trip—this single request might involve coordinating the capabilities of several specialized agents:

An agent for flight bookings.
An agent for hotel reservations.
A third agent for local tour recommendations and bookings.
A fourth agent handling currency conversion and travel advisories.

Without a common communication protocol, integrating these disparate agents into a unified user experience becomes a significant engineering challenge. Each integration would likely require a custom point-to-point solution, making the system difficult to scale, maintain, and extend.

Application Scenarios

Enterprise Automation

In corporate environments, A2A enables agents to work across siloed data systems and applications. For example, a supply chain planning agent can coordinate with inventory management, logistics, and procurement agents—even if they are built by different vendors or on different frameworks. This enhances autonomy, boosts productivity, and reduces long-term costs.

Multi-Agent Collaboration

A2A protocols facilitate true multi-agent scenarios where agents can collaborate in their natural, unstructured modes, even without shared memory, tools, or context. This goes beyond simply using one agent as a "tool" for another—it allows each agent to retain its capabilities while handling complex tasks.

Cross-Platform Integration

For business applications, A2A allows AI agents to operate across an entire ecosystem of enterprise applications. This means agents can access and coordinate with other agents on various platforms, such as CRM systems, knowledge bases, project management tools, and more. A standardized approach to managing agents across diverse platforms and cloud environments is essential for unlocking the full potential of collaborative AI.

How It Works

A2A facilitates communication between "client" agents and "remote" agents. The client agent is responsible for formulating and conveying tasks, while the remote agent executes them, aiming to provide accurate information or take appropriate actions. This interaction involves several key functions:

Capability Discovery: Agents can expose their capabilities using a JSON-formatted Agent Card. This allows client agents to identify the most suitable agent for a task and communicate with remote agents via A2A. For example, a client agent might discover another agent specialized in financial data and delegate financial analysis tasks to it.
Task Management: Communication between client and remote agents is task-oriented, with agents collaborating to fulfill end-user requests. The Task object, defined by the protocol, has a lifecycle. It can be completed immediately, or for long-running tasks, agents can synchronize updates. The output of a task is called an Artifact.
Collaboration: Agents can exchange messages to share context, responses, artifacts, or user instructions. This creates a structured way for agents to share the information needed to complete tasks. For instance, one agent might provide context about user preferences, while another returns analytical results.
Negotiation: Each message includes a parts field, which contains fully formed content fragments such as generated text or images. Each part has a specified content type, allowing client and remote agents to negotiate the correct format. It also explicitly supports negotiation for user UI features like iframes, videos, web forms, etc.

Core Concepts

Above, we outlined the basic working principles of A2A. This protocol involves a series of core concepts that define the interactions between agents.

Participants

First, we need to clarify the participants in the A2A protocol, including users, clients, and servers:

User: The end user (human or automated service) initiating a request or goal that requires assistance from an agent.
A2A Client (Client Agent): An application, service, or other AI agent that represents the user in requesting actions or information from a remote agent. The client initiates communication using the A2A protocol.
A2A Server (Remote Agent): An AI agent or agent system that exposes an HTTP endpoint implementing the A2A protocol. It receives requests from clients, processes tasks, and returns results or status updates. To the client, the remote agent operates as an "invisible" system, with no need for the client to understand its internal implementation.

Communication Elements

After identifying the participants, the next focus is the communication protocol—how these participants interact.

Agent Card

A JSON metadata document, typically discoverable via the URL /.well-known/agent.json, which describes the A2A server.
It details the agent's identity (name, description), service endpoint URLs, version, supported A2A features (e.g., streaming or push notifications), specific skills offered, default input/output methods, and authentication requirements.
Clients use the agent card to discover the agent and learn how to interact with it securely and effectively.

Task

When a client sends a message to an agent, the agent may determine that fulfilling the request requires completing a task (e.g., "generate a report," "book a flight," "answer a question").
Each task has a unique ID assigned by the agent and progresses through a defined lifecycle (e.g., submitted, working, input-required, completed, failed).
Tasks are stateful and may involve multiple exchanges (messages) between the client and server.

Message

Represents a single turn or unit of communication between the client and the agent.
A message has a role (messages sent by the client are labeled user, while those sent by the server are labeled agent) and contains one or more Part objects that carry the actual content. The messageId field is a unique identifier set by the sender for each message.
Used to convey instructions, context, questions, answers, or status updates, which may not necessarily be formal Artifacts.

Part

The fundamental content unit within a message or Artifact. Each part has a specific type and can carry different kinds of data:
- TextPart: Contains plain text content.
- FilePart: Represents a file, transmitted either as inline base64-encoded bytes or referenced via a URI. Includes metadata such as filename and media type.
- DataPart: Carries structured JSON data, suitable for forms, parameters, or any machine-readable information.

Artifact

Represents the output results generated by the remote agent during task processing.
Examples include generated documents, images, spreadsheets, structured data, or any other self-contained piece of information that directly results from the task.
An Artifact consists of one or more Part objects and can be streamed incrementally.

Interaction Mechanisms

Request/Response (Polling)

The client sends a request (e.g., using the message/send RPC method) and receives a response from the server.
If the interaction involves a stateful, long-running task, the server may initially respond with a working status. The client then periodically polls by calling tasks/get until the task reaches a terminal state (e.g., completed, failed).

Streaming (SSE)

Used for tasks that produce results incrementally or provide real-time progress updates.
The client initiates the interaction with the server using message/stream.
The server responds with an open HTTP connection, sending a stream of Server-Sent Events (SSE).
These events can be Task, Message, TaskStatusUpdateEvent (for state changes), or TaskArtifactUpdateEvent (for new or updated artifacts).
This requires the server to advertise streaming capability in its agent card.

Push Notifications

For very long-running tasks or scenarios where maintaining a persistent connection (e.g., SSE) is impractical.
The client may provide a webhook URL when initiating the task (or by calling tasks/pushNotificationConfig/set).
When significant task state changes occur (e.g., completed, failed, or input-required), the server can send an asynchronous notification (an HTTP POST request) to the client-provided webhook.
This requires the server to declare pushNotifications capability in its agent card.

Additional Concepts

Context (contextId): A server-generated identifier used to logically group related task objects, providing context for a series of interactions.
Transport and Format: A2A communication occurs over HTTP(S), with all request and response payloads formatted using JSON-RPC 2.0.
Authentication & Authorization: A2A relies on standard cybersecurity practices. Authentication requirements are declared in the agent card, and credentials (e.g., OAuth tokens, API keys) are typically passed via HTTP headers, separate from the A2A protocol messages themselves.
Agent Discovery: The process by which clients locate agent cards to identify available A2A servers and their capabilities.
Extensions: A2A allows agents to declare custom protocol extensions in their agent cards.

By understanding these core components and mechanisms, developers can effectively design, implement, and leverage A2A to build interoperable and collaborative AI agent systems.

User Case

Here is a real-world example of A2A in an enterprise scenario. A new employee is hired at a company, and multiple systems and departments are involved in the onboarding process:

HR needs to create records and send a welcome email.
The IT department needs to provide a laptop and company account.
The facilities team needs to prepare a desk and an access badge.

Traditionally, these steps were handled manually or through tightly coupled integrations between internal systems. Now, if each department exposes its own agent using the A2A protocol:

Agent	Responsibilities
`hr-agent.company.com`	Create employee records, send documents
`it-agent.company.com`	Set up email accounts, order laptops
`facilities-agent.company.com`	Assign desks, print badges

Suppose we have a multi-agent system, OnboardingPro (e.g., onboarding-agent.company.com), which orchestrates the entire onboarding workflow.

Discovery: It reads each agent's .well-known/agent.json to understand its capabilities and authentication.
Task Delegation:
- Sends the createEmployee task to the HR agent.
- Sends the setupEmailAccount and orderHardware tasks to the IT agent.
- Sends the assignDesk and generateBadge tasks to the facilities agent.
Continuous Updates: Agents stream back progress updates via server-sent events (e.g., "Laptop shipped," "Desk assigned").
Artifact Collection: Final outputs (e.g., PDF badge, confirmation email, account credentials) are returned as A2A artifacts.
Completion: OnboardingPro notifies the hiring manager that onboarding is complete.

A2A and MCP

As we have previously detailed in the MCP Protocol, some might assume that the A2A protocol is a replacement for MCP. However, this is not the case—the two are complementary. The MCP protocol primarily focuses on enabling AI models to dynamically access external data sources, invoke external tools, and obtain real-time information to execute tasks. If MCP is a socket wrench (for tools), then A2A is the conversation between mechanics (for collaboration). These two protocols perfectly complement each other in functionality. We recommend using MCP for tools and A2A for agents.

MCP: Focuses on connecting individual AI models with external tools and data sources (Model to Data/Tools).
A2A: Focuses on communication and collaboration between multiple AI agents (Agent to Agent).

An agent application might use A2A to communicate with other agents, while internally, each agent employs MCP to interact with its specific tools and resources.

A2A Protocol

After understanding some basic concepts of A2A, let's delve into how the specific A2A protocol is defined (version 0.2.2).

A2A communication must occur over HTTP(S), with all request and response payloads formatted using JSON-RPC 2.0. The A2A server provides services at the URL defined in its AgentCard.

Agent Discovery

The A2A server must provide an agent card. An agent card is a JSON document that describes the server's identity, capabilities, skills, service endpoint URLs, and how clients should authenticate and interact with it. Clients use this information to discover suitable agents and configure their interactions.

Clients can locate agent cards through several methods, including:

URI Strategy: Accessing a predefined path on the agent's domain. The recommended path for the agent card is https://{server_domain}/.well-known/agent.json, which is also the preferred approach.
Registry/Directory: An agent directory or registry.
Direct Configuration: Clients may pre-configure the URL of the agent card or the card content itself.

The agent card itself may contain sensitive information.

If the agent card includes sensitive information, the endpoint providing the card must be protected with appropriate access controls (e.g., mTLS, network restrictions, or requiring authentication to retrieve the card).
It is generally not recommended to include plaintext keys (such as static API keys) directly in the agent card.

AgentCard Object Structure

The structure of an AgentCard is as follows:

/**
 * An AgentCard conveys key information:
 * - General details (version, name, description, purpose)
 * - Skills: A set of capabilities the agent can perform
 * - Default modality/content type support
 * - Authentication requirements
 */
export interface AgentCard {
  /**
   * A human-readable name for the agent.
   * @example "Recipe Agent"
   */
  name: string;
  /**
   * A human-readable description of the agent. Helps users and other agents understand its capabilities.
   * @example "Agent that helps users with recipes and cooking."
   */
  description: string;
  /** The URL where the agent is hosted. */
  url: string;
  /** The URL of the agent's icon. */
  iconUrl?: string;
  /** The service provider of the agent. */
  provider?: AgentProvider;
  /**
   * The version of the agent—format determined by the provider.
   * @example "1.0.0"
   */
  version: string;
  /** The URL of the agent's documentation. */
  documentationUrl?: string;
  /** Optional capabilities supported by the agent. */
  capabilities: AgentCapabilities;
  /** Security scheme details for authentication. */
  securitySchemes?: { [scheme: string]: SecurityScheme };
  /** Security requirements for accessing the agent. */
  security?: { [scheme: string]: string[] }[];
  /**
   * The interaction modes supported by the agent for all skills. This can be overridden per skill.
   * Supported input media types.
   */
  defaultInputModes: string[];
  /** Supported output media types. */
  defaultOutputModes: string[];
  /** Skills represent units of capability the agent can perform. */
  skills: AgentSkill[];
  /**
   * If true, the agent supports providing an extended AgentCard upon user authentication.
   * Defaults to false if unspecified.
   */
  supportsAuthenticatedExtendedCard?: boolean;
}

Python SDK

In the previous section, we explored the A2A protocol in detail. Now, let's put it into practice by implementing it in specific scenarios using the official Python SDK.

Here, we'll use the Python SDK to build a simple "echo" A2A server, which will help us understand the fundamental concepts and components of an A2A server.

Initialization

First, ensure your Python version is 3.10 or higher. We'll continue using uv for project management.

Initialize the project:

uv init a2a-demo
cd a2a-demo

Then install the A2A Python SDK dependencies:

uv add a2a-sdk uvicorn

The project directory structure should now look like this:

$ tree .
.
├── README.md
├── main.py
├── pyproject.toml
└── uv.lock

0 directories, 4 files

Server Implementation

Next, we'll implement an A2A server (i.e., the remote agent) in the main.py file. We can leverage the components provided by a2a-sdk for this purpose. The core code is as follows:

import uvicorn

from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)
from agent_executor import (
    HelloWorldAgentExecutor,  # type: ignore[import-untyped]
)


if __name__ == '__main__':
    # Define a simple skill that returns "hello world"
    skill = AgentSkill(
        id='hello_world',
        name='Returns hello world',
        description='just returns hello world',
        tags=['hello world'],
        examples=['hi', 'hello world'],
    )

    # Define a more complex skill that returns "super hello world"
    extended_skill = AgentSkill(
        id='super_hello_world',
        name='Returns a SUPER Hello World',
        description='A more enthusiastic greeting, only for authenticated users.',
        tags=['hello world', 'super', 'extended'],
        examples=['super hi', 'give me a super hello'],
    )

    # Define a public agent card
    public_agent_card = AgentCard(
        name='Hello World Agent',
        description='Just a hello world agent',
        url='http://0.0.0.0:9999/',
        version='1.0.0',
        defaultInputModes=['text'],
        defaultOutputModes=['text'],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],  # Only basic skills for the public card
        supportsAuthenticatedExtendedCard=True,
    )

    # Define an extended agent card that returns "super hello world"
    specific_extended_agent_card = public_agent_card.model_copy(
        update={
            'name': 'Hello World Agent - Extended Edition',  # Different name
            'description': 'The full-featured hello world agent for authenticated users.',
            'version': '1.0.1',  # Can even be a different version
            # Capabilities and other fields like url, defaultInputModes, defaultOutputModes, supportsAuthenticatedExtendedCard, etc.,
            # are inherited from public_agent_card unless specified here.
            'skills': [
                skill,
                extended_skill,
            ],  # Both skills for the extended card
        }
    )

    # Create a default request handler
    request_handler = DefaultRequestHandler(
        agent_executor=HelloWorldAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    # Create an A2A server, here we use `A2AStarletteApplication` to create an A2A server
    server = A2AStarletteApplication(
        agent_card=public_agent_card,
        http_handler=request_handler,
        extended_agent_card=specific_extended_agent_card,
    )

    # Run the server, here we use `uvicorn` to run the server
    uvicorn.run(server.build(), host='0.0.0.0', port=9999)

First, we use the AgentSkill class to define a simple skill, which represents a unit of capability that an agent can perform. Then, we define a more complex skill.

Next, we use the AgentCard class to define an agent card, which represents the overall information of an agent. An AgentCard is used to convey key details:

General information (version, name, description, usage)
Skills: A set of capabilities the agent can perform
Default modality/content type support
Authentication requirements

The agent card we define here is as follows:

public_agent_card = AgentCard(
    name='Hello World Agent',
    description='Just a hello world agent',
    url='http://0.0.0.0:9999/',
    version='1.0.0',
    defaultInputModes=['text'],
    defaultOutputModes=['text'],
    capabilities=AgentCapabilities(streaming=True),
    skills=[skill],  # Only basic skills for the public card
    supportsAuthenticatedExtendedCard=True,
)

The agent information we define here includes:

name: The name of the agent
description: A description of the agent
url: The URL of the agent
version: The version of the agent
defaultInputModes: The input modes, here defined as text, indicating the agent supports text input
defaultOutputModes: The output modes, here defined as text, indicating the agent supports text output
capabilities: The agent's capabilities, here defined as streaming, indicating the agent supports streaming output
skills: The skills the agent supports, here defined as skill, representing the agent's supported skills
supportsAuthenticatedExtendedCard: Whether the agent supports an extended authenticated card

To extend the agent card, we can even override and create a new AgentCard, as shown below:

# Define an extended agent card to return "super hello world"
specific_extended_agent_card = public_agent_card.model_copy(
    update={
        'name': 'Hello World Agent - Extended Edition',  # Different name
        'description': 'The full-featured hello world agent for authenticated users.',
        'version': '1.0.1',  # Even a different version
        # Other fields like url, defaultInputModes, defaultOutputModes, supportsAuthenticatedExtendedCard, etc.,
        # unless specified here, will be inherited from public_agent_card.
        'skills': [
            skill,
            extended_skill,
        ],  # Two skills for the extended card
    }
)

Here, we provide two skill objects in the skills field, allowing us to use both skills in the extended agent card.

Next, we use the DefaultRequestHandler class provided by the SDK to create a default request handler, which is used to process A2A requests:

# Create a default request handler
request_handler = DefaultRequestHandler(
    agent_executor=HelloWorldAgentExecutor(),
    task_store=InMemoryTaskStore(),
)

This requires us to provide an instance of the AgentExecutor object, which is responsible for executing the core logic of the agent. We need to inherit from the AgentExecutor class ourselves. Additionally, we need to provide an instance of the TaskStore object, which is used to store tasks. Here, we use the InMemoryTaskStore class to implement this, indicating that tasks are stored in memory.

The key focus here is the AgentExecutor base class, whose basic definition is as follows:

from abc import ABC, abstractmethod

from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue


class AgentExecutor(ABC):
    """Agent executor interface.

    Classes implementing this interface contain the core logic of an agent, executing tasks based on requests and publishing updates to the event queue.
    """

    @abstractmethod
    async def execute(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        """Execute the agent's logic, performing tasks based on the request and publishing updates to the event queue.
        Args:
            context: Request context containing messages, task IDs, etc.
            event_queue: Queue for publishing events
        """

    @abstractmethod
    async def cancel(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        """Request the agent to cancel an ongoing task.
        Args:
            context: Request context containing the task ID to be canceled
            event_queue: Queue for publishing cancellation status update events
        """

The execute method is used to execute the agent's logic. The agent should read necessary information from the context and publish Task or Message events, or TaskStatusUpdateEvent / TaskArtifactUpdateEvent to the event queue. The event queue serves as a buffer for A2A to respond to the agent, bridging asynchronous execution and response handling (e.g., via SSE streaming).

The cancel method is used to cancel an ongoing task. The agent should attempt to stop the task identified in the context and publish a TaskStatusUpdateEvent with the status TaskState.canceled to the event queue.

Next, we create a file named agent_executor.py to implement the AgentExecutor interface, as shown below:

# agent_executor.py
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.utils import new_agent_text_message


class HelloWorldAgent:
    """Hello World agent."""
    async def invoke(self) -> str:
        return 'Hello World'


class HelloWorldAgentExecutor(AgentExecutor):
    """AgentExecutor implementation."""

    def __init__(self):
        self.agent = HelloWorldAgent()

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        # Execute the agent's logic—here, we directly return "Hello World"
        result = await self.agent.invoke()
        # Publish a text message to the event queue
        await event_queue.enqueue_event(new_agent_text_message(result))

    async def cancel(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        # Cancel logic for the agent—here, we raise an exception as cancellation is not supported
        raise Exception('cancel not supported')

As shown in the code above, we have implemented the AgentExecutor interface, including the execute and cancel methods. The execute method executes the agent's logic, while the cancel method cancels an ongoing task.

In the execute method, we directly call the invoke method of HelloWorldAgent, which represents the actual execution of the agent's logic. For example, we could integrate an LLM model, invoke it to perform the agent's logic, and then return the result (here, we return a fixed string). The result is then published to the event queue.

We use the new_agent_text_message function to convert the agent's execution result into a text message object, which corresponds to the Message object mentioned earlier in the protocol:

def new_agent_text_message(
    text: str,
    context_id: str | None = None,
    task_id: str | None = None,
) -> Message:
    """Create a new agent message containing a single TextPart.

    Args:
        text: The text content of the message
        context_id: The context ID of the message
        task_id: The task ID of the message

    Returns:
        A new `Message` object with the 'agent' role
    """
    return Message(
        role=Role.agent,
        parts=[Part(root=TextPart(text=text))],
        messageId=str(uuid.uuid4()),
        taskId=task_id,
        contextId=context_id,
    )

Once all preparations are complete, we use the A2AStarletteApplication class to create an A2A server and run it using uvicorn:

# Create an A2A server using `A2AStarletteApplication`
server = A2AStarletteApplication(
    agent_card=public_agent_card,
    http_handler=request_handler,
    extended_agent_card=specific_extended_agent_card,
)

# Run the server using `uvicorn`
uvicorn.run(server.build(), host='0.0.0.0', port=9999)

Finally, start the service:

$ python main.py
INFO:     Started server process [98132]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:9999 (Press CTRL+C to quit)

After running, you can access http://localhost:9999/.well-known/agent.json in a browser to retrieve the agent's card information.

Similarly, you can access http://localhost:9999/agent/authenticatedExtendedCard to retrieve the extended agent card information:

At this point, we have implemented the simplest A2A server program.

Client

Next, we will implement an A2A client to invoke the capabilities of remote agents using the agent card information from the server. Again, we will use the Python SDK to achieve this. The core code is as follows:

# client.py
# client.py
import logging

from typing import Any
from uuid import uuid4

import httpx

from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    MessageSendParams,
    SendMessageRequest,
    SendStreamingMessageRequest,
)


async def main() -> None:
    PUBLIC_AGENT_CARD_PATH = '/.well-known/agent.json'
    EXTENDED_AGENT_CARD_PATH = '/agent/authenticatedExtendedCard'

    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)

    base_url = 'http://localhost:9999'

    async with httpx.AsyncClient() as httpx_client:
        # Initialize A2ACardResolver
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=base_url,
            # agent_card_path uses default value, extended_agent_card_path also uses default value
        )

        # Fetch public agent card and initialize the client
        final_agent_card_to_use: AgentCard | None = None

        try:
            logger.info(
                f'Attempting to fetch public agent card from: {base_url}{PUBLIC_AGENT_CARD_PATH}'
            )
            _public_card = (
                await resolver.get_agent_card()
            )  # Fetch from the default public path
            logger.info('Successfully fetched public agent card:')
            logger.info(
                _public_card.model_dump_json(indent=2, exclude_none=True)
            )
            final_agent_card_to_use = _public_card
            logger.info(
                '\nUsing PUBLIC agent card for client initialization (default).'
            )

            if _public_card.supportsAuthenticatedExtendedCard:
                try:
                    logger.info(
                        f'\nPublic card supports authenticated extended card. Attempting to fetch from: {base_url}{EXTENDED_AGENT_CARD_PATH}'
                    )
                    auth_headers_dict = {
                        'Authorization': 'Bearer dummy-token-for-extended-card'
                    }
                    _extended_card = await resolver.get_agent_card(
                        relative_card_path=EXTENDED_AGENT_CARD_PATH,
                        http_kwargs={'headers': auth_headers_dict},
                    )
                    logger.info(
                        'Successfully fetched authenticated extended agent card:'
                    )
                    logger.info(
                        _extended_card.model_dump_json(
                            indent=2, exclude_none=True
                        )
                    )
                    final_agent_card_to_use = (
                        _extended_card  # Update to use the extended agent card
                    )
                    logger.info(
                        '\nUsing AUTHENTICATED EXTENDED agent card for client initialization.'
                    )
                except Exception as e_extended:
                    logger.warning(
                        f'Failed to fetch extended agent card: {e_extended}. Will proceed with public card.',
                        exc_info=True,
                    )
            elif (
                _public_card
            ):  # supportsAuthenticatedExtendedCard is False or None
                logger.info(
                    '\nPublic card does not indicate support for an extended card. Using public card.'
                )

        except Exception as e:
            logger.error(
                f'Critical error fetching public agent card: {e}', exc_info=True
            )
            raise RuntimeError(
                'Failed to fetch the public agent card. Cannot continue.'
            ) from e

        # Initialize A2AClient object
        client = A2AClient(
            httpx_client=httpx_client, agent_card=final_agent_card_to_use
        )
        logger.info('A2AClient initialized.')

        send_message_payload: dict[str, Any] = {
            'message': {
                'role': 'user',
                'parts': [
                    {'kind': 'text', 'text': 'how much is 10 USD in INR?'}
                ],
                'messageId': uuid4().hex,
            },
        }
        # Construct a send message request object
        request = SendMessageRequest(
            id=str(uuid4()), params=MessageSendParams(**send_message_payload)
        )
        # Send the message
        response = await client.send_message(request)
        # Print the response
        print(response.model_dump(mode='json', exclude_none=True))

        # Construct a streaming send message request object
        streaming_request = SendStreamingMessageRequest(
            id=str(uuid4()), params=MessageSendParams(**send_message_payload)
        )

        # Send the streaming message
        stream_response = client.send_message_streaming(streaming_request)

        # Print the streaming message response
        async for chunk in stream_response:
            print(chunk.model_dump(mode='json', exclude_none=True))


if __name__ == '__main__':
    import asyncio

    asyncio.run(main())

First, we request the agent card information from the A2A server (via the /.well-known/agent.json path). Then, we check whether the server supports providing an extended agent card upon user authentication. If supported, we request the extended agent card information (via the /agent/authenticatedExtendedCard path).

Once we have the agent card data, we only need to initialize an A2AClient object to start invoking the agent's capabilities. For example, we can construct a send message request object SendMessageRequest and then call the send_message method to send the message and receive the remote agent's response. To construct a streaming message request, we create a SendStreamingMessageRequest object and call the send_message_streaming method to send the message and receive the streaming response data.

Finally, we run the client program directly for testing:

$ python client.py
INFO:__main__:Attempting to fetch public agent card from: http://localhost:9999/.well-known/agent.json
INFO:httpx:HTTP Request: GET http://localhost:9999/.well-known/agent.json "HTTP/1.1 200 OK"
INFO:a2a.client.client:Successfully fetched agent card data from http://localhost:9999/.well-known/agent.json: {'capabilities': {'streaming': True}, 'defaultInputModes': ['text'], 'defaultOutputModes': ['text'], 'description': 'Just a hello world agent', 'name': 'Hello World Agent', 'skills': [{'description': 'just returns hello world', 'examples': ['hi', 'hello world'], 'id': 'hello_world', 'name': 'Returns hello world', 'tags': ['hello world']}], 'supportsAuthenticatedExtendedCard': True, 'url': 'http://0.0.0.0:9999/', 'version': '1.0.0'}
INFO:__main__:Successfully fetched public agent card:
INFO:__main__:{
  "capabilities": {
    "streaming": true
  },
  "defaultInputModes": [
    "text"
  ],
  "defaultOutputModes": [
    "text"
  ],
  "description": "Just a hello world agent",
  "name": "Hello World Agent",
  "skills": [
    {
      "description": "just returns hello world",
      "examples": [
        "hi",
        "hello world"
      ],
      "id": "hello_world",
      "name": "Returns hello world",
      "tags": [
        "hello world"
      ]
    }
  ],
  "supportsAuthenticatedExtendedCard": true,
  "url": "http://0.0.0.0:9999/",
  "version": "1.0.0"
}
INFO:__main__:
Using PUBLIC agent card for client initialization (default).
INFO:__main__:
Public card supports authenticated extended card. Attempting to fetch from: http://localhost:9999/agent/authenticatedExtendedCard
INFO:httpx:HTTP Request: GET http://localhost:9999/agent/authenticatedExtendedCard "HTTP/1.1 200 OK"
INFO:a2a.client.client:Successfully fetched agent card data from http://localhost:9999/agent/authenticatedExtendedCard: {'capabilities': {'streaming': True}, 'defaultInputModes': ['text'], 'defaultOutputModes': ['text'], 'description': 'The full-featured hello world agent for authenticated users.', 'name': 'Hello World Agent - Extended Edition', 'skills': [{'description': 'just returns hello world', 'examples': ['hi', 'hello world'], 'id': 'hello_world', 'name': 'Returns hello world', 'tags': ['hello world']}, {'description': 'A more enthusiastic greeting, only for authenticated users.', 'examples': ['super hi', 'give me a super hello'], 'id': 'super_hello_world', 'name': 'Returns a SUPER Hello World', 'tags': ['hello world', 'super', 'extended']}], 'supportsAuthenticatedExtendedCard': True, 'url': 'http://0.0.0.0:9999/', 'version': '1.0.1'}
INFO:__main__:Successfully fetched authenticated extended agent card:
INFO:__main__:{
  "capabilities": {
    "streaming": true
  },
  "defaultInputModes": [
    "text"
  ],
  "defaultOutputModes": [
    "text"
  ],
  "description": "The full-featured hello world agent for authenticated users.",
  "name": "Hello World Agent - Extended Edition",
  "skills": [
    {
      "description": "just returns hello world",
      "examples": [
        "hi",
        "hello world"
      ],
      "id": "hello_world",
      "name": "Returns hello world",
      "tags": [
        "hello world"
      ]
    },
    {
      "description": "A more enthusiastic greeting, only for authenticated users.",
      "examples": [
        "super hi",
        "give me a super hello"
      ],
      "id": "super_hello_world",
      "name": "Returns a SUPER Hello World",
      "tags": [
        "hello world",
        "super",
        "extended"
      ]
    }
  ],
  "supportsAuthenticatedExtendedCard": true,
  "url": "http://0.0.0.0:9999/",
  "version": "1.0.1"
}
INFO:__main__:
Using AUTHENTICATED EXTENDED agent card for client initialization.
INFO:__main__:A2AClient initialized.
INFO:httpx:HTTP Request: POST http://0.0.0.0:9999/ "HTTP/1.1 200 OK"
{'id': 'c8c2b71f-89bf-4a31-a660-0c6a1b50f7ea', 'jsonrpc': '2.0', 'result': {'kind': 'message', 'messageId': 'b50b5c2d-10b9-4800-80a4-4f38f25f1556', 'parts': [{'kind': 'text', 'text': 'Hello World'}], 'role': 'agent'}}
INFO:httpx:HTTP Request: POST http://0.0.0.0:9999/ "HTTP/1.1 200 OK"
{'id': 'a2a6316d-7aef-4b51-b7d1-535bbc1362eb', 'jsonrpc': '2.0', 'result': {'kind': 'message', 'messageId': 'f6904b90-7e3a-4596-b5e8-5c14920443f6', 'parts': [{'kind': 'text', 'text': 'Hello World'}], 'role': 'agent'}}

In the output above, we first requested the public agent card information, then fetched the extended agent card information, and finally sent a message to the remote agent, receiving its response.

At this point, we have implemented the simplest A2A server and client program.

Here, we have merely demonstrated the basic mechanics of A2A. In real-world scenarios, our agents would be more sophisticated, supporting advanced features such as streaming, task state management, and LLM-driven multi-turn conversations. Different agents might also be implemented using various frameworks. We could even integrate MCP within an agent to enhance functionality. Furthermore, we could develop an aggregation platform where users could configure the card addresses provided by various agents, enabling them to invoke the capabilities of these agents through the platform. These more complex functionalities will be covered in subsequent articles.

What is Manus AI?

cnych — Sun, 06 Jul 2025 01:50:00 +0000

Manus AI: Redefining the Future of AI Assistants

In today's rapidly evolving AI landscape, we've witnessed countless groundbreaking technological advancements. From simple chatbots to sophisticated language models, AI technology is transforming how we work and live at an unprecedented pace. Yet in this field brimming with innovation, one name is quietly rising—not just representing technological progress, but symbolizing a new paradigm in human-computer interaction. This is Manus AI.

Manus AI is not just another AI assistant. It is the world's first truly general artificial intelligence agent, an intelligent partner that can directly translate your thoughts into actions. When we discuss Manus AI, we're actually talking about a significant milestone in AI development history—a revolutionary product that may fundamentally change how we interact with technology.

What is Manus AI?

The name Manus AI derives from the Latin word for "hand" (Manus), and this naming is no coincidence. Just as human hands are our primary tools for interacting with the physical world, Manus AI is designed to be the "intelligent hand" for users to interact with the digital world. It not only understands user needs but, more importantly, can proactively execute tasks, truly achieving "thought-to-action."

In traditional AI assistant concepts, we're accustomed to receiving suggestions, answers, or guidance. But Manus AI breaks this passive interaction model by taking on the role of an executor. When you present Manus AI with a complex task requirement, it doesn't simply tell you what to do—it systematically plans the entire execution process, calls upon necessary tools and resources, and completes the entire task for you.

This shift from "advisor" to "executor" represents a major leap in AI applications. It means we no longer need to manually execute tasks after receiving AI suggestions but can delegate entire tasks to AI, much like assigning work to an experienced colleague.

Core Technical Advantages of Manus AI

1. Exceptional Benchmark Performance

When discussing Manus AI's technical capabilities, we must highlight its outstanding performance in the GAIA (General AI Assistant) benchmark. GAIA is currently the industry's most challenging evaluation standard for AI assistants, testing not just AI's comprehension but, more critically, its ability to execute tasks in complex real-world scenarios.

In this rigorous test, the human average score is 92%, already a remarkably high standard. Yet even the most advanced GPT-4 with plugin systems achieves only 15%. This vast gap clearly demonstrates the limitations of current AI technology in practical task execution.

But Manus AI achieves SOTA (State-of-the-Art) performance in this test, far surpassing other AI assistants across all three difficulty levels. This isn't just a numerical victory but strong validation of Manus AI's technical architecture and design philosophy. It proves Manus AI genuinely possesses the capability to independently complete tasks in complex real-world environments.

2. Revolutionary Tool Invocation Capability

Traditional chatbots facing user requests can typically only provide text-based answers or suggestions. Even the most advanced language models are mainly limited to information processing and generation, unable to directly interact with external systems. This limitation means users must still manually perform specific operations after receiving AI suggestions.

Manus AI fundamentally changes this. It possesses powerful tool invocation capabilities, enabling proactive interaction with various external systems, applications, and services. Whether accessing databases, manipulating file systems, calling API interfaces, or controlling various software tools, Manus AI handles them all with ease.

Implementing this capability requires complex technical architecture support. Manus AI must not only understand user intent but also translate it into concrete execution steps, then invoke appropriate tools to complete each step. This process involves task decomposition, tool selection, parameter configuration, execution monitoring, and more—each requiring highly intelligent processing.

3. Independent Operating Environment

Another key feature of Manus AI is its independent computing environment. This means it can work autonomously without requiring continuous user supervision, much like a true digital employee.

This independence manifests in multiple ways. First, Manus AI can run continuously in the background, handling lengthy complex tasks without needing users to stay online or monitor progress. Second, it can autonomously manage resources, including computing power, storage, and network connections. Most importantly, it can make independent decisions during execution, handling unexpected situations and anomalies.

Achieving this autonomous operation capability requires robust task management and resource scheduling systems. Manus AI must prioritize tasks appropriately, allocate computing resources efficiently, and independently seek solutions when problems arise. This capability makes Manus AI not just a tool but a reliable work partner.

A New Paradigm for Human-AI Interaction

One of Manus AI's most exciting features is its creation of a new human-AI interaction paradigm. Traditional AI interaction is essentially "Q&A-style": users ask questions, AI provides answers; users make requests, AI offers suggestions. While useful, this remains fundamentally a "tool-like" interaction.

Manus AI changes this. Interacting with Manus AI is more like collaborating with an experienced colleague. You describe your goals and expectations, and it proactively takes responsibility for achieving them. Throughout the process, it maintains communication, reports progress, seeks input, and even suggests improvements.

This collaborative interaction model offers several key advantages. First, it significantly reduces users' cognitive load. Users don't need detailed knowledge of every execution step—just clear expression of goals. Second, it improves efficiency. Since AI can handle multiple tasks in parallel without fatigue or emotional impact, overall productivity sees marked improvement.

Most importantly, this model makes AI a true extension of human capability rather than just an external tool. Users can view Manus AI as a team member—always reliable, always online, always proactive.

Practical Applications of Manus AI

Complex Data Analysis and Report Generation

In modern business environments, data analysis is both crucial and complex. Traditional workflows typically involve data collection, cleaning, analysis, visualization, and reporting—each step requiring specialized skills and significant time investment.

Manus AI can autonomously complete the entire data analysis process. Simply tell it what data to analyze and what insights you seek, and it will automatically gather data from various sources, perform necessary cleaning and preprocessing, run appropriate analytical algorithms, generate intuitive visualizations, and produce a comprehensive analysis report.

This capability's value lies not just in efficiency gains but in uncovering data patterns and correlations humans might miss. Manus AI can process multi-dimensional data simultaneously, run complex statistical models, and never tire or err from large datasets.

Automated Content Creation and Management

Content creation is another area where Manus AI excels. Whether drafting technical documents, creating marketing materials, or generating social media content, Manus AI can produce high-quality output tailored to user requirements and brand style.

More importantly, Manus AI doesn't just create content—it manages entire production workflows. It can schedule creation tasks based on content calendars, ensure consistency and quality, and even adjust content strategies based on audience feedback.

This automated content management capability offers tremendous value for businesses requiring large-scale content output, significantly reducing production costs while ensuring timeliness and relevance.

Intelligent Project Management and Coordination

Project management demands high-level coordination skills. Traditional approaches require constant progress tracking, resource coordination, conflict resolution, and plan adjustments—time-consuming, labor-intensive, and prone to human error.

Manus AI can serve as an intelligent project management assistant, automatically tracking progress, identifying potential risks and bottlenecks, coordinating team schedules, and even proposing optimization suggestions.

With its ability to handle multiple projects simultaneously and powerful data analysis skills, Manus AI can learn from historical project data to continuously improve management strategies, enhancing performance over time.

How to Use Manus AI Effectively

Clear Requirement Expression

To fully leverage Manus AI's capabilities, users must learn effective communication. The key principle is clarity and specificity. When presenting tasks, describe expectations in detail—including objectives, quality standards, timelines, etc.

For example, instead of "analyze sales data," say "analyze the past six months' sales data, focusing on performance differences across product lines, with improvement suggestions. The report should include visualizations and actionable recommendations." Such detailed descriptions help Manus AI better understand needs and deliver superior results.

Progressive Task Decomposition

For particularly complex tasks, adopt a progressive approach. Rather than describing all requirements at once, have Manus AI complete portions first, then provide further guidance based on intermediate results.

This method allows course correction during execution, ensuring final outcomes meet expectations while helping users better understand Manus AI's capabilities for more reasonable task planning.

Personalized Knowledge Systems

Manus AI learns and remembers. Through interaction, users can gradually build personalized knowledge systems. When satisfied with a task's outcome, instruct it to "remember this approach for similar future tasks."

This personalized learning mechanism helps Manus AI increasingly adapt to users' work styles and preferences, delivering ever more precise service over time.

Timely Feedback and Adjustment

If execution deviates from expectations during a task, provide immediate feedback. Don't wait until completion to adjust, as this wastes time and resources.

Manus AI can quickly adapt strategies based on real-time feedback—a flexibility traditional automation systems lack and a key advantage.

Manus AI vs. Traditional AI Assistants

To better understand Manus AI's unique value, let's compare it to traditional AI assistants:

In tool invocation, traditional chatbots typically have limited or no such capability. Even advanced AI systems often restrict tool use to specific scenarios. Manus AI features advanced, systematic tool invocation, seamlessly integrating with various external systems.

In work style, traditional AI assistants require continuous supervision and guidance, while Manus AI operates independently, usually without constant oversight.

In interaction style, traditional AI feels like using a complex tool requiring learning to "operate" properly. Manus AI interaction resembles human colleague collaboration—more natural and intuitive.

These differences represent not just technical improvements but a fundamental shift in AI application philosophy. Manus AI aims not to replace humans but to become an intelligent extension of human capability, helping us better achieve our goals.

The AI Agent Ecosystem Development

Manus AI's emergence isn't isolated but part of the broader AI Agent ecosystem. Other notable projects include:

TARS Agent—ByteDance's open-source multimodal AI Agent client excelling in workflow orchestration and real-time artifact generation. OpenManus—an open-source general AI agent framework offering flexible capability configuration and easy tool integration. OWL—a multi-agent collaboration framework based on CAMEL-AI, also performing excellently in GAIA benchmarks.

These projects indicate rapid AI Agent technology development industry-wide toward smarter, more autonomous solutions. Manus AI holds a unique position—not just leading technically but setting new standards in user experience and practicality.

Future Outlook

Manus AI marks a pivotal moment in AI development—transitioning from "AI as tool" to "AI as partner." In this new era, AI no longer passively awaits instructions but actively participates in human work and life as a true intelligent assistant.

As technology advances, we can expect Manus AI to demonstrate value across more domains—from simple task automation to complex decision support, personal productivity to enterprise-level digital transformation.

More importantly, Manus AI's success will inspire more researchers and developers to advance AI Agent technology, accelerating industry growth. We can confidently predict that soon, intelligent agents like Manus AI will become indispensable partners in our daily work and lives.

Conclusion

Manus AI isn't just a technological product—it represents a new direction and possibility in AI development. By translating thoughts into actions, Manus AI is redefining human-computer interaction boundaries and pioneering a new era of intelligent work.

For anyone interested in AI development, Manus AI warrants deep exploration. It not only enhances work efficiency but reveals AI's true potential and future trajectory.

In this era of change and opportunity, Manus AI is writing AI's next chapter in its own way. Let's anticipate how, under Manus AI's leadership, artificial intelligence will bring more convenience and possibilities to our work and lives.

Manus AI is now open for registration. Sign up via https://manus.im/invitation/39HXFEFJBCFOM to receive 1500 credits + 300 daily credits bonus.

Developing an SSE-based Claude MCP shopping ai assistant service

cnych — Wed, 02 Apr 2025 12:58:43 +0000

Claude MCP supports two communication transport methods: STDIO (Standard Input/Output) or SSE (Server-Sent Events), both using JSON-RPC 2.0 for message formatting. STDIO is used for local integration, while SSE is used for network-based communication.

For example, if you need to use an MCP service directly in the command line, the STDIO transport method is appropriate. For web-based applications, the SSE transport method is the better choice.

In this article, we'll build an intelligent shopping assistant using an SSE-type MCP service with these key capabilities:

Real-time product information and inventory access with custom order support
Personalized product recommendations based on customer preferences and inventory availability
Seamless interaction with microservices through the MCP tool server
Live inventory checking when responding to product inquiries
Streamlined product purchasing using product IDs and quantities
Instant inventory updates after transactions
Natural language analysis of order data

We'll use Anthropic Claude 3.5 Sonnet as our AI assistant for this MCP service, though any model supporting tool calling would work.

Our implementation requires several components:

A product microservice exposing API endpoints for product information
An order microservice with APIs for order creation and inventory management
An MCP SSE server that exposes microservice data to the LLM as tools via the SSE protocol
An MCP client that connects to the server through SSE to interact with the LLM

The complete project code is available at https://github.com/cnych/mcp-sse-demo

Microservices

Let's start by developing our product and order microservices with their API interfaces.

First, we'll define the types for products, inventory, and orders.

// types/index.ts
export interface Product {
  id: number;
  name: string;
  price: number;
  description: string;
}

export interface Inventory {
  productId: number;
  quantity: number;
  product?: Product;
}

export interface Order {
  id: number;
  customerName: string;
  items: Array<{ productId: number; quantity: number }>;
  totalAmount: number;
  orderDate: string;
}

Then we can use Express to expose the product and order microservices and provide API interfaces. Since it's a simulation, we'll use simpler in-memory data here, and expose the data through the following functions. (In a production environment, you still need to use a microservice plus a database to implement it.)

// services/product-service.ts
import { Product, Inventory, Order } from "../types/index.js";

// Simulate data storage
let products: Product[] = [
  {
    id: 1,
    name: "智能手表Galaxy",
    price: 1299,
    description: "健康监测，运动追踪，支持多种应用",
  },
  {
    id: 2,
    name: "无线蓝牙耳机Pro",
    price: 899,
    description: "主动降噪，30小时续航，IPX7防水",
  },
  {
    id: 3,
    name: "便携式移动电源",
    price: 299,
    description: "20000mAh大容量，支持快充，轻薄设计",
  },
  {
    id: 4,
    name: "华为MateBook X Pro",
    price: 1599,
    description: "14.2英寸全面屏，3:2比例，100% sRGB色域",
  },
];

// Simulate inventory data
let inventory: Inventory[] = [
  { productId: 1, quantity: 100 },
  { productId: 2, quantity: 50 },
  { productId: 3, quantity: 200 },
  { productId: 4, quantity: 150 },
];

let orders: Order[] = [];

export async function getProducts(): Promise<Product[]> {
  return products;
}

export async function getInventory(): Promise<Inventory[]> {
  return inventory.map((item) => {
    const product = products.find((p) => p.id === item.productId);
    return {
      ...item,
      product,
    };
  });
}

export async function getOrders(): Promise<Order[]> {
  return [...orders].sort(
    (a, b) => new Date(b.orderDate).getTime() - new Date(a.orderDate).getTime()
  );
}

export async function createPurchase(
  customerName: string,
  items: { productId: number; quantity: number }[]
): Promise<Order> {
  if (!customerName || !items || items.length === 0) {
    throw new Error("Invalid request: missing customer name or items");
  }

  let totalAmount = 0;

  // Verify inventory and calculate total price
  for (const item of items) {
    const inventoryItem = inventory.find((i) => i.productId === item.productId);
    const product = products.find((p) => p.id === item.productId);

    if (!inventoryItem || !product) {
      throw new Error(`Product ID ${item.productId} does not exist`);
    }

    if (inventoryItem.quantity < item.quantity) {
      throw new Error(
        `Product ${product.name} has insufficient inventory. Available: ${inventoryItem.quantity}`
      );
    }

    totalAmount += product.price * item.quantity;
  }

  // Create order
  const order: Order = {
    id: orders.length + 1,
    customerName,
    items,
    totalAmount,
    orderDate: new Date().toISOString(),
  };

  // Update inventory
  items.forEach((item) => {
    const inventoryItem = inventory.find(
      (i) => i.productId === item.productId
    )!;
    inventoryItem.quantity -= item.quantity;
  });

  orders.push(order);
  return order;
}

Then we can expose these API interfaces through MCP tools, as shown below:

// mcp-server.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import {
  getProducts,
  getInventory,
  getOrders,
  createPurchase,
} from "./services/product-service.js";

export const server = new McpServer({
  name: "mcp-sse-demo",
  version: "1.0.0",
  description:
    "MCP tools for product query, inventory management, and order processing",
});

// Get product list tool
server.tool("getProducts", "Get all product information", {}, async () => {
  console.log("Getting product list");
  const products = await getProducts();
  return {
    content: [
      {
        type: "text",
        text: JSON.stringify(products),
      },
    ],
  };
});

// Get inventory information tool
server.tool(
  "getInventory",
  "Get all product inventory information",
  {},
  async () => {
    console.log("Getting inventory information");
    const inventory = await getInventory();
    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(inventory),
        },
      ],
    };
  }
);

// Get order list tool
server.tool("getOrders", "Get all order information", {}, async () => {
  console.log("Getting order list");
  const orders = await getOrders();
  return {
    content: [
      {
        type: "text",
        text: JSON.stringify(orders),
      },
    ],
  };
});

// Purchase product tool
server.tool(
  "purchase",
  "Purchase product",
  {
    items: z
      .array(
        z.object({
          productId: z.number().describe("Product ID"),
          quantity: z.number().describe("Purchase quantity"),
        })
      )
      .describe("List of products to purchase"),
    customerName: z.string().describe("Customer name"),
  },
  async ({ items, customerName }) => {
    console.log("Processing purchase request", { items, customerName });
    try {
      const order = await createPurchase(customerName, items);
      return {
        content: [
          {
            type: "text",
            text: JSON.stringify(order),
          },
        ],
      };
    } catch (error: any) {
      return {
        content: [
          {
            type: "text",
            text: JSON.stringify({ error: error.message }),
          },
        ],
      };
    }
  }
);

Here we define 4 tools:

getProducts: Retrieve all product information
getInventory: Get inventory information for all products
getOrders: Retrieve all order information
purchase: Purchase products

If it's a Stdio-type MCP service, we can use these tools directly in the command line, but now we need to use an SSE-type MCP service, so we also need an MCP SSE server to expose these tools.

MCP SSE Server

Next, we'll develop an MCP SSE server to expose the product and order microservice data as tools using the SSE protocol.

// mcp-sse-server.ts
import express, { Request, Response, NextFunction } from "express";
import cors from "cors";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
import { server as mcpServer } from "./mcp-server.js"; // Rename to avoid naming conflict

const app = express();
app.use(
  cors({
    origin: process.env.ALLOWED_ORIGINS?.split(",") || "*",
    methods: ["GET", "POST"],
    allowedHeaders: ["Content-Type", "Authorization"],
  })
);

// Store active connections
const connections = new Map();

// Health check endpoint
app.get("/health", (req, res) => {
  res.status(200).json({
    status: "ok",
    version: "1.0.0",
    uptime: process.uptime(),
    timestamp: new Date().toISOString(),
    connections: connections.size,
  });
});

// SSE connection establishment endpoint
app.get("/sse", async (req, res) => {
  // Instantiate SSE transport object
  const transport = new SSEServerTransport("/messages", res);
  // Get sessionId
  const sessionId = transport.sessionId;
  console.log(
    `[${new Date().toISOString()}] New SSE connection established: ${sessionId}`
  );

  // Register connection
  connections.set(sessionId, transport);

  // Connection interruption handling
  req.on("close", () => {
    console.log(
      `[${new Date().toISOString()}] SSE connection closed: ${sessionId}`
    );
    connections.delete(sessionId);
  });

  // Connect the transport object to the MCP server
  await mcpServer.connect(transport);
  console.log(
    `[${new Date().toISOString()}] MCP server connection successful: ${sessionId}`
  );
});

// Endpoint for receiving client messages
app.post("/messages", async (req: Request, res: Response) => {
  try {
    console.log(
      `[${new Date().toISOString()}] Received client message:`,
      req.query
    );
    const sessionId = req.query.sessionId as string;

    // Find the corresponding SSE connection and process the message
    if (connections.size > 0) {
      const transport: SSEServerTransport = connections.get(
        sessionId
      ) as SSEServerTransport;
      // Use transport to process messages
      if (transport) {
        await transport.handlePostMessage(req, res);
      } else {
        throw new Error("No active SSE connection");
      }
    } else {
      throw new Error("No active SSE connection");
    }
  } catch (error: any) {
    console.error(
      `[${new Date().toISOString()}] Failed to process client message:`,
      error
    );
    res
      .status(500)
      .json({ error: "Failed to process message", message: error.message });
  }
});

// Graceful shutdown of all connections
async function closeAllConnections() {
  console.log(
    `[${new Date().toISOString()}] Closing all connections (${
      connections.size
    }个)`
  );
  for (const [id, transport] of connections.entries()) {
    try {
      // Send shutdown event
      transport.res.write(
        'event: server_shutdown\ndata: {"reason": "Server is shutting down"}\n\n'
      );
      transport.res.end();
      console.log(`[${new Date().toISOString()}] Connection closed: ${id}`);
    } catch (error) {
      console.error(
        `[${new Date().toISOString()}] Failed to close connection: ${id}`,
        error
      );
    }
  }
  connections.clear();
}

// Error handling
app.use((err: Error, req: Request, res: Response, next: NextFunction) => {
  console.error(`[${new Date().toISOString()}] Unhandled exception:`, err);
  res.status(500).json({ error: "Server internal error" });
});

// Graceful shutdown
process.on("SIGTERM", async () => {
  console.log(
    `[${new Date().toISOString()}] Received SIGTERM signal, preparing to close`
  );
  await closeAllConnections();
  server.close(() => {
    console.log(`[${new Date().toISOString()}] Server closed`);
    process.exit(0);
  });
});

process.on("SIGINT", async () => {
  console.log(
    `[${new Date().toISOString()}] Received SIGINT signal, preparing to close`
  );
  await closeAllConnections();
  process.exit(0);
});

// Start server
const port = process.env.PORT || 8083;
const server = app.listen(port, () => {
  console.log(
    `[${new Date().toISOString()}] Smart shopping MCP SSE server started, address: http://localhost:${port}`
  );
  console.log(`- SSE connection endpoint: http://localhost:${port}/sse`);
  console.log(
    `- Message processing endpoint: http://localhost:${port}/messages`
  );
  console.log(`- Health check endpoint: http://localhost:${port}/health`);
});

Here we use Express to expose an SSE connection endpoint /sse, for receiving client messages. We use SSEServerTransport to create an SSE transport object and specify the message processing endpoint as /messages.

const transport = new SSEServerTransport("/messages", res);

After the transport object is created, we can connect the transport object to the MCP server, as shown below:

// Connect the transport object to the MCP server
await mcpServer.connect(transport);

Then we can receive client messages through the SSE connection endpoint /sse, and use the message processing endpoint /messages to process client messages. When a client message is received, we need to use the transport object to handle the client message in the /messages endpoint:

// Use transport to process messages
await transport.handlePostMessage(req, res);

This is the same as the operation of listing tools and calling tools.

MCP Client

Next, we'll develop an MCP client to connect to the MCP SSE server and interact with the LLM. We can develop a command line client or a web client.

For the command line client, we've already introduced it, the only difference is that now we need to use the SSE protocol to connect to the MCP SSE server.

// Create MCP client
const mcpClient = new McpClient({
  name: "mcp-sse-demo",
  version: "1.0.0",
});

// Create SSE transport object
const transport = new SSEClientTransport(new URL(config.mcp.serverUrl));

// Connect to MCP server
await mcpClient.connect(transport);

After establishing the connection, the workflow mirrors our previous command line client implementation: we enumerate available tools, forward the user's query along with tool definitions to the LLM, process any tool invocations from the LLM response, and finally send the tool execution results back to the LLM to generate the comprehensive answer.

For our web-based implementation, the core logic remains identical to the command line version, with the key difference being that we now expose these processing steps through API endpoints that our web frontend can interact with.

The initialization process begins with setting up the MCP client, retrieving the available tools, transforming them into Anthropic's expected format (as their API requires a specific schema structure), and finally instantiating the Anthropic client that will handle our AI interactions.

// Initialize MCP client
async function initMcpClient() {
  if (mcpClient) return;

  try {
    console.log("Connecting to MCP server...");
    mcpClient = new McpClient({
      name: "mcp-client",
      version: "1.0.0",
    });

    const transport = new SSEClientTransport(new URL(config.mcp.serverUrl));

    await mcpClient.connect(transport);
    const { tools } = await mcpClient.listTools();
    // Convert tool format to the array form required by Anthropic
    anthropicTools = tools.map((tool: any) => {
      return {
        name: tool.name,
        description: tool.description,
        input_schema: tool.inputSchema,
      };
    });
    // Create Anthropic client
    aiClient = createAnthropicClient(config);

    console.log("MCP client and tools initialized");
  } catch (error) {
    console.error("Failed to initialize MCP client:", error);
    throw error;
  }
}

Then we can develop API interfaces according to our own needs, for example, we develop a chat interface here to receive the user's question, then call the MCP client's tools, send the tool call results and history messages back to the LLM for processing to get the final result, as shown below:

// API: Chat request
apiRouter.post("/chat", async (req, res) => {
  try {
    const { message, history = [] } = req.body;

    if (!message) {
      console.warn("请求中消息为空");
      return res.status(400).json({ error: "消息不能为空" });
    }

    // Build message history
    const messages = [...history, { role: "user", content: message }];

    // Call AI
    const response = await aiClient.messages.create({
      model: config.ai.defaultModel,
      messages,
      tools: anthropicTools,
      max_tokens: 1000,
    });

    // Process tool calls
    const hasToolUse = response.content.some(
      (item) => item.type === "tool_use"
    );

    if (hasToolUse) {
      // Process all tool calls
      const toolResults = [];

      for (const content of response.content) {
        if (content.type === "tool_use") {
          const name = content.name;
          const toolInput = content.input as
            | { [x: string]: unknown }
            | undefined;

          try {
            // Call MCP tool
            if (!mcpClient) {
              console.error("MCP client not initialized");
              throw new Error("MCP client not initialized");
            }
            console.log(`Calling MCP tool: ${name}`);
            const toolResult = await mcpClient.callTool({
              name,
              arguments: toolInput,
            });

            toolResults.push({
              name,
              result: toolResult,
            });
          } catch (error: any) {
            console.error(`Tool call failed: ${name}`, error);
            toolResults.push({
              name,
              error: error.message,
            });
          }
        }
      }

      // Send tool results back to AI to get the final response
      const finalResponse = await aiClient.messages.create({
        model: config.ai.defaultModel,
        messages: [
          ...messages,
          {
            role: "user",
            content: JSON.stringify(toolResults),
          },
        ],
        max_tokens: 1000,
      });

      const textResponse = finalResponse.content
        .filter((c) => c.type === "text")
        .map((c) => c.text)
        .join("\n");

      res.json({
        response: textResponse,
        toolCalls: toolResults,
      });
    } else {
      // Return AI response directly
      const textResponse = response.content
        .filter((c) => c.type === "text")
        .map((c) => c.text)
        .join("\n");

      res.json({
        response: textResponse,
        toolCalls: [],
      });
    }
  } catch (error: any) {
    console.error("Chat request processing failed:", error);
    res.status(500).json({ error: error.message });
  }
});

The core implementation is also relatively simple, and is basically the same as the command line client, except that now we implement these processing steps in some interfaces.

Usage

Here is an example of using the command line client:

We can also use it in Cursor, create a .cursor/mcp.json file, and add the following content:

{
  "mcpServers": {
    "products-sse": {
      "url": "http://localhost:8083/sse"
    }
  }
}

Then we can see this MCP service in the Cursor settings page, and then we can use this MCP service in Cursor.

Here is an example of using the web client we developed:

Summary

When implementing tool calling functionality with LLMs, the effectiveness largely depends on well-crafted tool descriptions:

Clear Documentation: Each tool should have precise, unambiguous descriptions with strategic keywords that help the model determine appropriate usage contexts
Functional Distinction: Tools should have clearly differentiated purposes to prevent the LLM from making incorrect selections between similar options
Comprehensive Testing: Before deployment, validate tool calling accuracy across diverse user query patterns to ensure reliable performance

The MCP server architecture offers implementation flexibility through multiple technology options:

Python SDK implementation
TypeScript/JavaScript frameworks
Various other programming language options

Technology selection should align with your team's expertise and complement your existing infrastructure.

Incorporating an AI assistant with MCP capabilities into a microservice ecosystem delivers several strategic benefits:

Dynamic Data Delivery: SSE protocol enables immediate data transmission, critical for time-sensitive information like inventory levels and order tracking
Independent Scaling: System components can be scaled according to their specific demand patterns, optimizing resource allocation
Fault Isolation: The distributed architecture prevents cascading failures, as individual service disruptions remain contained
Technical Diversity: Teams can develop and maintain different system components using their preferred technology stacks
Bandwidth Optimization: SSE's event-driven approach reduces network overhead compared to polling methods by transmitting only when changes occur
Responsive Interfaces: Immediate updates and rapid system responses significantly enhance the customer experience
Streamlined Frontend: Client implementations become more elegant without complex polling logic, simply subscribing to server events

For production deployments, several additional considerations become important:

Implement rigorous error detection through comprehensive testing protocols
Establish robust failover mechanisms to maintain service continuity
Deploy performance monitoring systems that track tool calling accuracy and response times
Evaluate caching strategies to minimize backend service load during peak usage

By following these architectural principles, you can create a high-performance, reliable MCP-powered shopping assistant that delivers personalized, real-time experiences to your customers.

Origin article link: https://www.claudemcp.com/docs/dev-sse-mcp