Начало работы с MCP, ADK и A2A

Начало работы с MCP, ADK и A2A

О практической работе

subjectПоследнее обновление: июн. 25, 2025
account_circleАвторы: Jack Wotherspoon

1. Обзор

Агенты ИИ стремительно набирают популярность, совершая революцию в автоматизации задач и принятии решений благодаря своей способности действовать автономно, обучаться и взаимодействовать с окружающей средой для достижения целей.

Но как именно создать агента? Эта кодовая лаборатория поможет вам начать, показав, как создать валютного агента, который может конвертировать валюты разных стран. С целью пройтись по новейшим технологиям, чтобы помочь вам понять те аббревиатуры, которые вы, возможно, видели в Интернете (MCP, ADK, A2A).

Архитектура

Модель контекстного протокола (MCP)

Model Context Protocol (MCP) — открытый протокол, который стандартизирует, как приложения предоставляют контекст LLM. MCP обеспечивает стандартизированный способ подключения моделей ИИ к ресурсам, подсказкам и инструментам.

Комплект для разработки агента (ADK)

Agent Development Kit (ADK) — это гибкая структура оркестровки для разработки и развертывания агентов ИИ. ADK не зависит от модели, развертывания и создан для совместимости с другими структурами. ADK был разработан, чтобы сделать разработку агентов более похожей на разработку программного обеспечения, чтобы разработчикам было проще создавать, развертывать и оркестровать архитектуры агентов, которые варьируются от простых задач до сложных рабочих процессов.

Протокол Агент2Агент (A2A)

Протокол Agent2Agent (A2A) — это открытый стандарт, разработанный для обеспечения бесперебойной коммуникации и сотрудничества между агентами ИИ. Так же, как MCP предоставляет стандартизированный способ предоставления LLM доступа к данным и инструментам, A2A предоставляет стандартизированный способ для агентов общаться с другими агентами! В мире, где агенты создаются с использованием различных фреймворков и разными поставщиками, A2A предоставляет общий язык, разрушая разрозненность и способствуя взаимодействию.

Чему вы научитесь

  • Как создать локальный MCP-сервер
  • Развертывание сервера MCP в Cloud Run
  • Как создать агента с помощью Agent Development Kit, использующего инструменты MCP
  • Как представить агента ADK как сервер A2A
  • Тестирование сервера A2A с использованием клиента A2A

Что вам понадобится

  • Браузер, например Chrome или Firefox
  • Проект Google Cloud с включенной оплатой.

2. Прежде чем начать

Создать проект

  1. В Google Cloud Console на странице выбора проекта выберите или создайте проект Google Cloud.
  2. Убедитесь, что для вашего проекта Cloud включена функция выставления счетов. Узнайте, как проверить, включена ли функция выставления счетов для проекта .
  3. Активируйте Cloud Shell, нажав на эту ссылку . Вы можете переключаться между Cloud Shell Terminal (для запуска облачных команд) и Editor (для создания проектов), нажав на соответствующую кнопку в Cloud Shell.
  4. После подключения к Cloud Shell вы проверяете, что вы уже аутентифицированы и что проекту присвоен ваш идентификатор проекта, с помощью следующей команды: 
gcloud auth list
  1. Выполните следующую команду в Cloud Shell, чтобы подтвердить, что команда gcloud знает о вашем проекте. 
gcloud config list project
  1. Для настройки проекта используйте следующую команду: 
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
  1. Включите требуемые API с помощью следующей команды. Это может занять несколько минут. 
gcloud services enable cloudresourcemanager.googleapis.com \
                       servicenetworking.googleapis.com \
                       run.googleapis.com \
                       cloudbuild.googleapis.com \
                       artifactregistry.googleapis.com \
                       aiplatform.googleapis.com \
                       compute.googleapis.com
  1. Убедитесь, что у вас установлен Python 3.10+.

Информацию о командах и использовании gcloud см. в документации .

3. Установка

  1. Клонируйте репозиторий: 
git clone https://github.com/jackwotherspoon/currency-agent.git
cd currency-agent
  1. Установка uv (используется для управления зависимостями): 
# macOS and Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (uncomment below line)
# powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
  1. Настройте переменные среды (через файл .env ):

Создайте файл .env , выполнив следующую команду:

echo "GOOGLE_GENAI_USE_VERTEXAI=TRUE" >> .env \
&& echo "GOOGLE_CLOUD_PROJECT=$PROJECT_ID" >> .env \
&& echo "GOOGLE_CLOUD_LOCATION=us-central1" >> .env

4. Создать локальный MCP-сервер

Прежде чем приступить к организации работы вашего валютного агента, вам сначала необходимо создать сервер MCP для предоставления инструментов, которые понадобятся вашему агенту.

Сервер MCP позволяет вам писать легкие программы для предоставления определенных возможностей (например, получения курсов обмена валют) в качестве инструментов. Агент или даже несколько агентов могут затем получить доступ к этим инструментам с помощью стандартизированного протокола контекста модели (MCP).

Пакет FastMCP Python можно использовать для создания сервера MCP, который предоставляет один инструмент под названием get_exchange_rate . Инструмент get_exchange_rate делает вызов через интернет к Frankfurter API, чтобы получить текущий обменный курс между двумя валютами.

Код для сервера MCP можно найти в файле mcp-server/server.py :

import logging
import os

import httpx
from fastmcp import FastMCP

# Set up logging
logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)

mcp = FastMCP("Currency MCP Server 💵")

@mcp.tool()
def get_exchange_rate(
    currency_from: str = 'USD',
    currency_to: str = 'EUR',
    currency_date: str = 'latest',
):
    """Use this to get current exchange rate.

    Args:
        currency_from: The currency to convert from (e.g., "USD").
        currency_to: The currency to convert to (e.g., "EUR").
        currency_date: The date for the exchange rate or "latest". Defaults to "latest".

    Returns:
        A dictionary containing the exchange rate data, or an error message if the request fails.
    """
    logger.info(f"--- 🛠️ Tool: get_exchange_rate called for converting {currency_from} to {currency_to} ---")
    try:
        response = httpx.get(
            f'https://api.frankfurter.app/{currency_date}',
            params={'from': currency_from, 'to': currency_to},
        )
        response.raise_for_status()

        data = response.json()
        if 'rates' not in data:
            return {'error': 'Invalid API response format.'}
        logger.info(f'✅ API response: {data}')
        return data
    except httpx.HTTPError as e:
        return {'error': f'API request failed: {e}'}
    except ValueError:
        return {'error': 'Invalid JSON response from API.'}

if __name__ == "__main__":
    logger.info(f"🚀 MCP server started on port {os.getenv('PORT', 8080)}")
    # Could also use 'sse' transport, host="0.0.0.0" required for Cloud Run.
    asyncio.run(
        mcp.run_async(
            transport="streamable-http",
            host="0.0.0.0",
            port=os.getenv("PORT", 8080),
        )
    )

Чтобы запустить сервер MCP локально, откройте терминал и выполните следующую команду (сервер запустится по http://localhost:8080 ):

uv run mcp-server/server.py

Проверьте правильность работы сервера MCP и доступность инструмента get_exchange_rate с использованием протокола контекста модели.

В новом окне терминала (чтобы не останавливать локальный сервер MCP) выполните следующее: 

uv run mcp-server/test_server.py

Вы должны увидеть текущий обменный курс 1 USD (доллар США) к EUR (евро): 

--- 🛠️ Tool found: get_exchange_rate ---
--- 🪛 Calling get_exchange_rate tool for USD to EUR ---
--- ✅ Success: {
  "amount": 1.0,
  "base": "USD",
  "date": "2025-05-26",
  "rates": {
    "EUR": 0.87866
  }
} ---

Отлично! У вас успешно работает сервер MCP с инструментом, к которому ваш агент сможет получить доступ.

Прежде чем перейти к следующей станции, остановите локально запущенный сервер MCP, нажав Ctrl+C (или Command+C на Mac) в терминале, где вы его запустили.

5. Разверните свой сервер MCP в облаке Run

Теперь вы готовы развернуть сервер MCP как удаленный сервер MCP в Cloud Run 🚀☁️

Преимущества удаленного запуска сервера MCP

Удалённый запуск сервера MCP в Cloud Run может обеспечить ряд преимуществ:

  • 📈Масштабируемость : Cloud Run создан для быстрого масштабирования для обработки всех входящих запросов . Cloud Run будет масштабировать ваш сервер MCP автоматически в зависимости от спроса.
  • 👥Централизованный сервер : вы можете предоставить доступ к централизованному серверу MCP членам команды через привилегии IAM, что позволит им подключаться к нему со своих локальных машин вместо того, чтобы запускать свои собственные серверы локально. Если в сервер MCP будут внесены изменения, все члены команды получат от этого выгоду.
  • 🔐Безопасность : Cloud Run предоставляет простой способ принудительного выполнения аутентифицированных запросов. Это позволяет устанавливать только безопасные соединения с вашим сервером MCP, предотвращая несанкционированный доступ.

Перейдите в каталог mcp-server : 

cd mcp-server

Разверните сервер MCP в Cloud Run: 

gcloud run deploy mcp-server --no-allow-unauthenticated --region=us-central1 --source .

Если ваша служба успешно развернута, вы увидите сообщение следующего вида: 

Service [mcp-server] revision [mcp-server-12345-abc] has been deployed and is serving 100 percent of traffic.

Аутентификация клиентов MCP

Поскольку вы указали --no-allow-unauthenticated для требования аутентификации, любому клиенту MCP, подключающемуся к удаленному серверу MCP, необходимо будет пройти аутентификацию.

Официальная документация по серверам Host MCP в Cloud Run содержит дополнительную информацию по этой теме в зависимости от того, где вы запускаете свой клиент MCP.

Вам потребуется запустить прокси-сервер Cloud Run , чтобы создать аутентифицированный туннель к удаленному серверу MCP на вашем локальном компьютере.

По умолчанию URL-адрес сервисов Cloud Run требует, чтобы все запросы были авторизованы с ролью IAM Cloud Run Invoker ( roles/run.invoker ). Эта привязка политики IAM гарантирует, что для аутентификации вашего локального клиента MCP используется надежный механизм безопасности.

Вам следует убедиться, что у вас или у любого члена команды, пытающегося получить доступ к удаленному серверу MCP, IAM-роль roles/run.invoker привязана к их принципалу IAM (аккаунту Google Cloud).

gcloud run services proxy mcp-server --region=us-central1

Вы должны увидеть следующий вывод: 

Proxying to Cloud Run service [mcp-server] in project [<YOUR_PROJECT_ID>] region [us-central1]
http://127.0.0.1:8080 proxies to https://mcp-server-abcdefgh-uc.a.run.app

Весь трафик на http://127.0.0.1:8080 теперь будет аутентифицироваться и перенаправляться на удаленный сервер MCP.

Тестирование удаленного сервера MCP

В новом терминале вернитесь в корневую папку и повторно запустите файл mcp-server/test_server.py чтобы убедиться, что удаленный сервер MCP работает.

cd ..
uv run mcp-server/test_server.py

Вы должны увидеть такой же вывод, как и при локальном запуске сервера: 

--- 🛠️ Tool found: get_exchange_rate ---
--- 🪛 Calling get_exchange_rate tool for USD to EUR ---
--- ✅ Success: {
  "amount": 1.0,
  "base": "USD",
  "date": "2025-05-26",
  "rates": {
    "EUR": 0.87866
  }
} ---

Вы можете запросить журналы развернутого сервера Cloud Run MCP, если хотите убедиться, что удаленный сервер действительно был вызван: 

gcloud run services logs read mcp-server --region us-central1 --limit 5

В журналах вы должны увидеть следующее: 

2025-06-04 14:28:29,871 [INFO]: --- 🛠️ Tool: get_exchange_rate called for converting USD to EUR ---
2025-06-04 14:28:30,610 [INFO]: HTTP Request: GET https://api.frankfurter.app/latest?from=USD&to=EUR "HTTP/1.1 200 OK"
2025-06-04 14:28:30,611 [INFO]: ✅ API response: {'amount': 1.0, 'base': 'USD', 'date': '2025-06-03', 'rates': {'EUR': 0.87827}}

Теперь, когда у вас есть удаленный сервер MCP, вы можете перейти к созданию агента! 🤖

6. Создайте агента с помощью Agent Development Kit (ADK)

У вас есть развернутый сервер MCP, теперь пришло время создать валютный агент с помощью Agent Development Kit (ADK) .

Agent Development Kit недавно выпустил свой стабильный релиз v1.0.0. Этот этап означает, что Python ADK теперь готов к производству, предлагая разработчикам надежную и устойчивую платформу для уверенного создания и развертывания своих агентов в реальных средах.

ADK делает создание агентов чрезвычайно легким и позволяет им легко подключаться к серверам MCP со встроенной поддержкой MCP Tools. Валютный агент будет получать доступ к инструменту get_exchange_rate , используя класс MCPToolset ADK.

Код валютного агента находится в currency_agent/agent.py : 

import logging
import os

from dotenv import load_dotenv
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import MCPToolset, StreamableHTTPConnectionParams

logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)

load_dotenv()

SYSTEM_INSTRUCTION = (
    "You are a specialized assistant for currency conversions. "
    "Your sole purpose is to use the 'get_exchange_rate' tool to answer questions about currency exchange rates. "
    "If the user asks about anything other than currency conversion or exchange rates, "
    "politely state that you cannot help with that topic and can only assist with currency-related queries. "
    "Do not attempt to answer unrelated questions or use tools for other purposes."
)

def create_agent() -> LlmAgent:
    """Constructs the ADK currency conversion agent."""
    logger.info("--- 🔧 Loading MCP tools from MCP Server... ---")
    logger.info("--- 🤖 Creating ADK Currency Agent... ---")
    return LlmAgent(
        model="gemini-2.5-flash",
        name="currency_agent",
        description="An agent that can help with currency conversions",
        instruction=SYSTEM_INSTRUCTION,
        tools=[
            MCPToolset(
                connection_params=StreamableHTTPConnectionParams(
                    url=os.getenv("MCP_SERVER_URL", "http://localhost:8080/mcp")
                )
            )
        ],
    )


root_agent = create_agent()

Чтобы быстро протестировать валютного агента, вы можете воспользоваться пользовательским интерфейсом разработчика ADK, доступ к которому можно получить, запустив adk web : 

uv run adk web

В браузере перейдите по адресу http://localhost:8000 чтобы увидеть и протестировать агента!

Убедитесь, что в левом верхнем углу веб-интерфейса в качестве агента выбран currency_agent .

Веб-интерфейс ADK

Задайте своему агенту в чате вопрос вроде «Сколько будет 250 CAD в USD?» . Вы должны увидеть, как агент вызывает наш инструмент MCP get_exchange_rate прежде чем он даст ответ.

Агент по обмену валют ADK Web

Агент работает! Он может обрабатывать запросы, связанные с конвертацией валют 💸.

7. Протокол Агент2Агент (A2A)

Протокол Agent2Agent (A2A) — это открытый стандарт, разработанный для обеспечения бесперебойной коммуникации и сотрудничества между агентами ИИ. Это позволяет агентам, созданным с использованием различных фреймворков и разными поставщиками, общаться друг с другом на общем языке, разрушая разрозненность и способствуя взаимодействию.

Протокол А2А

A2A позволяет агентам:

  • Откройте для себя: найдите других агентов и изучите их навыки ( AgentSkill ) и возможности ( AgentCapabilities ), используя стандартизированные карты агентов .
  • Общайтесь: обменивайтесь сообщениями и данными безопасно.
  • Сотрудничайте: делегируйте задачи и координируйте действия для достижения сложных целей.

Протокол A2A упрощает эту коммуникацию с помощью таких механизмов, как «агентские карты», которые действуют как цифровые визитные карточки, используемые агентами для рекламы своих возможностей и информации о подключении.

Карта агента A2A

Теперь пришло время раскрыть валютного агента с помощью A2A, чтобы к нему могли обращаться другие агенты и клиенты.

A2A Python SDK

A2A Python SDK предоставляет модели Pydantic для каждого из вышеупомянутых ресурсов: AgentSkill , AgentCapabilities и AgentCard . Это обеспечивает интерфейс для ускорения разработки и интеграции с протоколом A2A.

AgentSkill — это то, как вы будете рекламировать другим агентам, что у валютного агента есть инструмент для get_exchange_rate : 

# A2A Agent Skill definition
skill = AgentSkill(
    id='get_exchange_rate',
    name='Currency Exchange Rates Tool',
    description='Helps with exchange values between various currencies',
    tags=['currency conversion', 'currency exchange'],
    examples=['What is exchange rate between USD and GBP?'],
)

Затем в AgentCard будут перечислены навыки и возможности агента, а также дополнительные сведения, такие как режимы ввода и вывода, с которыми может работать агент: 

# A2A Agent Card definition
agent_card = AgentCard(
    name='Currency Agent',
    description='Helps with exchange rates for currencies',
    url=f'http://{host}:{port}/',
    version='1.0.0',
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    capabilities=AgentCapabilities(streaming=True),
    skills=[skill],
)

Интерфейс AgentExecutor обрабатывает основную логику того, как агент A2A обрабатывает запросы и генерирует ответы/события. A2A Python SDK предоставляет абстрактный базовый класс a2a.server.agent_execution.AgentExecutor , который вам нужно реализовать.

Пришло время объединить все это с валютным агентом и продемонстрировать мощь A2A!

8. Сервер валютного агента A2A

Теперь вы взглянете на некоторые фрагменты кода и увидите, как объединяются различные части, составляющие сервер A2A.

Заглянув в файл currency_agent/agent_executor.py , вы увидите, что у вас есть класс ADKAgentExecutor , который наследуется от абстрактного класса AgentExecutor A2A. Он обрабатывает вызов агента ADK, вызывая бегуна ADK, обрабатывая запросы к агенту и выполняя преобразование между google.genai.types , которые использует ADK, и a2a.types , которые использует A2A. 

# ... see file for full code

class ADKAgentExecutor(AgentExecutor):
    """An AgentExecutor that runs an ADK agent."""

    def __init__(self, runner: Runner, card: AgentCard):
        self.runner = runner
        self._card = card
        self._running_sessions = {}

    def _run_agent(
        self, session_id, new_message: types.Content
    ) -> AsyncGenerator[Event, None]:
        return self.runner.run_async(
            session_id=session_id, user_id="self", new_message=new_message
        )

    async def _process_request(
        self,
        new_message: types.Content,
        session_id: str,
        task_updater: TaskUpdater,
    ) -> None:
        session = await self._upsert_session(
            session_id,
        )
        session_id = session.id
        # Run through all events within the request.
        async for event in self._run_agent(session_id, new_message):
            if event.is_final_response():
                parts = convert_genai_parts_to_a2a(event.content.parts)
                logger.debug("✅ Yielding final response: %s", parts)
                await task_updater.add_artifact(parts)
                await task_updater.complete()
                break
            # If the agent is not making a function call, yield an update.
            if not event.get_function_calls():
                logger.debug("⏳ Yielding update response")
                await task_updater.update_status(
                    TaskState.working,
                    message=task_updater.new_agent_message(
                        convert_genai_parts_to_a2a(event.content.parts),
                    ),
                )
            else:
                logger.debug("➡️ Skipping event")

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ):
        # Run the agent until either complete or the task is suspended.
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        # Immediately notify that the task is submitted.
        if not context.current_task:
            updater.submit()
        updater.start_work()
        await self._process_request(
            types.UserContent(
                parts=convert_a2a_parts_to_genai(context.message.parts),
            ),
            context.context_id,
            updater,
        )
        logger.debug("--- 💵💱💶 [Currency] execute exiting ---")

# ... see file for full code

Внутри currency_agent/__main__.py вы инициализируете AgentSkill, AgentCard и создаете валютного агента ADK. Там же вы настраиваете и запускаете сервер A2A .

A2A Python SDK предоставляет класс A2AFastAPIApplication , который упрощает запуск HTTP-сервера, совместимого с A2A. Он использует FastAPI для веб-фреймворка и обычно запускается с сервером ASGI, таким как Uvicorn . 

# ... see file for full code
@click.command()
@click.option("--host", "host", default="localhost")
@click.option("--port", "port", default=10000)
def main(host: str, port: int):
    # Verify one of Google AI Studio or Vertex AI is being used
    if os.getenv("GOOGLE_GENAI_USE_VERTEXAI") != "TRUE" and not os.getenv(
        "GOOGLE_API_KEY"
    ):
        raise ValueError(
            "GOOGLE_API_KEY environment variable not set and "
            "GOOGLE_GENAI_USE_VERTEXAI is not TRUE."
        )

    # A2A Agent Skill definition
    skill = AgentSkill(
        id="get_exchange_rate",
        name="Currency Exchange Rates Tool",
        description="Helps with exchange values between various currencies",
        tags=["currency conversion", "currency exchange"],
        examples=["What is exchange rate between USD and GBP?"],
    )

    # A2A Agent Card definition
    agent_card = AgentCard(
        name="Currency Agent",
        description="Helps with exchange rates for currencies",
        url=f"http://{host}:{port}/",
        version="1.0.0",
        defaultInputModes=["text"],
        defaultOutputModes=["text"],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],
    )

    # Create the ADK runner and executor.
    runner = Runner(
        app_name=agent_card.name,
        agent=root_agent,
        artifact_service=InMemoryArtifactService(),
        session_service=InMemorySessionService(),
        memory_service=InMemoryMemoryService(),
    )
    agent_executor = ADKAgentExecutor(runner, agent_card)

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

    server = A2AFastAPIApplication(
        agent_card=agent_card, http_handler=request_handler
    )

    uvicorn.run(server.build(), host=host, port=port)
# ... see file for full code

Чтобы запустить сервер A2A, в новом терминале выполните следующее: 

uv run currency_agent/

Если сервер запустится успешно, вывод будет выглядеть следующим образом, указывая на то, что он работает на порту 10000 : 

[INFO]: --- 🔧 Loading MCP tools from MCP Server... ---
[INFO]: --- 🤖 Creating ADK Currency Agent... ---
INFO:     Started server process [45824]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:10000 (Press CTRL+C to quit)

Валютный агент теперь успешно работает как сервер A2A, и к нему могут обращаться другие агенты или клиенты, использующие протокол A2A!

Тест сервера A2A

Теперь вы можете протестировать сервер, отправив ему несколько запросов с помощью A2A!

A2A Python SDK предоставляет класс a2a.client.A2AClient , который упрощает эту задачу.

Файл currency_agent/test_client.py содержит код, который проходит несколько различных тестовых случаев на сервере A2A. 

# ... see file for full code

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

    send_message_payload = create_send_message_payload(text="how much is 100 USD in CAD?")
    request = SendMessageRequest(
        id=str(uuid4()), params=MessageSendParams(**send_message_payload)
    )

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

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

    task_id: str = 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")

# ----- Main Entrypoint (Create client --> Run tests) -----
async def main() -> None:
    """Main function to run the tests."""
    print(f'--- 🔄 Connecting to agent at {AGENT_URL}... ---')
    try:
        async with httpx.AsyncClient() 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)
            await run_streaming_test(client)
            await run_multi_turn_test(client)

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

Запустите тесты с помощью следующей команды: 

uv run currency_agent/test_client.py

Успешный тестовый запуск приведет к следующему: 

--- 🔄 Connecting to agent at http://localhost:10000... ---
--- ✅ Connection successful. ---
--- ✉️ Single Turn Request ---
--- 📥 Single Turn Request Response ---
{"id":"3bc92d7b-d857-4e93-9ff0-b2fb865f6e35","jsonrpc":"2.0","result":{"artifacts":[{"artifactId":"35e89e14-b977-4397-a23b-92c84bc32379","parts":[{"kind":"text","text":"Based on the current exchange rate, 1 USD is equivalent to 1.3704 CAD. Therefore, 100 USD would be 137.04 CAD.\n"}]}],"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","history":[{"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","kind":"message","messageId":"59819269f7d04849b0bfca7d43ec073c","parts":[{"kind":"text","text":"how much is 100 USD in CAD?"}],"role":"user","taskId":"52ae2392-84f5-429a-a14b-8413d3d20d97"},{"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","kind":"message","messageId":"286095c6-12c9-40cb-9596-a9676d570dbd","parts":[],"role":"agent","taskId":"52ae2392-84f5-429a-a14b-8413d3d20d97"}],"id":"52ae2392-84f5-429a-a14b-8413d3d20d97","kind":"task","status":{"state":"completed"}}}

// ...

--- ⏩ Single Turn Streaming Request ---
--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"state":"submitted"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}

--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"state":"working"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}

--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"message":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","kind":"message","messageId":"25f5f972-9475-4e4a-a08d-e13f521d7462","parts":[],"role":"agent","taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"},"state":"working"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}

--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"artifact":{"artifactId":"35e89e14-b977-4397-a23b-92c84bc32379","parts":[{"kind":"text","text":"The current exchange rate is 1 EUR to 164.15 JPY. So, 50 EUR would be 8207.5 JPY.\n"}]},"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","kind":"artifact-update","taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}

// ...

--- 🚀 First turn completed, no further input required for this test case. ---

Это работает! Вы успешно проверили, что можете общаться с валютным агентом через сервер A2A! 🎉

Посетите репозиторий a2a-samples на GitHub, чтобы увидеть более продвинутые варианты использования!

Хотите развернуть своего агента? Vertex AI Agent Engine предоставляет управляемый опыт для развертывания агентов ИИ в производстве!

9. Поздравления

Поздравляем! Вы успешно создали и развернули удаленный сервер MCP, создали агента валюты с помощью Agent Development Kit (ADK), который подключается к инструментам с помощью MCP, и выставили своего агента с помощью протокола Agent2Agent (A2A)! Теперь агент валюты доступен для взаимодействия с другими агентами любой платформы с помощью A2A!

Вот ссылка на полную документацию кода.

Что мы рассмотрели

  • Как создать локальный MCP-сервер
  • Развертывание сервера MCP в Cloud Run
  • Как создать агента с помощью Agent Development Kit, использующего инструменты MCP
  • Как представить агента ADK как сервер A2A
  • Тестирование сервера A2A с использованием клиента A2A

Очистить

Чтобы избежать списания средств с вашего аккаунта Google Cloud за ресурсы, используемые в этой лабораторной работе, выполните следующие действия:

  1. В консоли Google Cloud перейдите на страницу Управление ресурсами .
  2. В списке проектов выберите проект, который вы хотите удалить, и нажмите Удалить .
  3. В диалоговом окне введите идентификатор проекта, а затем нажмите «Завершить работу» , чтобы удалить проект.
bug_report Сообщить об ошибке