Informacje o tym ćwiczeniu (w Codelabs)
1. Przegląd
Agenci AI zyskują coraz większą popularność, rewolucjonizując automatyzację zadań i podejmowanie decyzji dzięki możliwości samodzielnego działania, uczenia się i interakcji z otoczeniem w celu osiągania celów.
Jak dokładnie tworzy się agenta? Ten przewodnik pomoże Ci zacząć. Dowiesz się z niego, jak utworzyć agenta walutowego, który będzie mógł przeliczać waluty różnych krajów. Chcemy zapoznać Cię z najnowszymi technologiami, aby pomóc Ci zrozumieć akronimy, które możesz widzieć w internecie (MCP, ADK, A2A).
Model Context Protocol (MCP)
Protokół kontekstu modelu (MCP) to otwarty protokół, który standaryzuje sposób, w jaki aplikacje dostarczają kontekst do dużych modeli językowych. MCP to standardowy sposób łączenia modeli AI z zasobami, promptami i narzędziami.
Pakiet do tworzenia agentów (ADK)
Pakiet Agent Development Kit (ADK) to elastyczna platforma do tworzenia i wdrażania agentów AI. ADK jest niezależny od modelu i wdrożenia oraz został stworzony z myślą o kompatybilności z innymi platformami. Pakiet ADK został zaprojektowany tak, aby tworzenie agentów przypominało tworzenie oprogramowania. Ułatwia on deweloperom tworzenie, wdrażanie i orkiestrację architektur agentów, które obejmują zarówno proste zadania, jak i złożone procesy.
Protokół Agent2Agent (A2A)
Protokół Agent2Agent (A2A) to otwarty standard zaprojektowany z myślą o umożliwieniu płynnej komunikacji i współpracy między agentami AI. Podobnie jak MCP zapewnia standardowy sposób na przyznawanie dużym modelom językowym dostępu do danych i narzędzi, A2A zapewnia standardowy sposób na komunikację agentów z innymi agentami. W świecie, w którym agenci są tworzeni przy użyciu różnych platform i przez różnych dostawców, A2A zapewnia wspólny język, który eliminuje bariery i wspiera interoperacyjność.
Czego się nauczysz
- Jak utworzyć lokalny serwer MCP
- Wdrażanie serwera MCP w Cloud Run
- Jak utworzyć agenta za pomocą pakietu ADK, który korzysta z narzędzi MCP
- Udostępnianie agenta ADK jako serwera A2A
- Testowanie serwera A2A za pomocą klienta A2A
Czego potrzebujesz
2. Zanim zaczniesz
Utwórz projekt
- W konsoli Google Cloud na stronie selektora projektów wybierz lub utwórz projekt Google Cloud.
- Sprawdź, czy w projekcie Cloud włączone są płatności. Dowiedz się, jak sprawdzić, czy w projekcie włączone są płatności.
- Aktywuj Cloud Shell, klikając ten link. Możesz przełączać się między terminalem Cloud Shell (do uruchamiania poleceń w chmurze) a edytorem (do tworzenia projektów), klikając odpowiedni przycisk w Cloud Shell.
- Po połączeniu z Cloud Shell sprawdź, czy jesteś już uwierzytelniony i czy projekt jest ustawiony na Twój identyfikator projektu, używając tego polecenia:
gcloud auth list
- Aby potwierdzić, że polecenie gcloud zna Twój projekt, uruchom w Cloud Shell to polecenie:
gcloud config list project
- Aby ustawić projekt, użyj tego polecenia:
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
- Włącz wymagane interfejsy API za pomocą tego polecenia. Może to potrwać kilka minut.
gcloud services enable cloudresourcemanager.googleapis.com \
servicenetworking.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
aiplatform.googleapis.com \
compute.googleapis.com
- Upewnij się, że masz Pythona w wersji 3.10 lub nowszej.
Informacje o poleceniach gcloud i ich użyciu znajdziesz w dokumentacji.
3. Instalacja
- Sklonuj repozytorium:
git clone https://github.com/jackwotherspoon/currency-agent.git
cd currency-agent
- Zainstaluj uv (do zarządzania zależnościami):
# 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"
- Skonfiguruj zmienne środowiskowe (za pomocą pliku
.env):
Utwórz plik .env, uruchamiając to polecenie:
echo "GOOGLE_GENAI_USE_VERTEXAI=TRUE" >> .env \
&& echo "GOOGLE_CLOUD_PROJECT=$PROJECT_ID" >> .env \
&& echo "GOOGLE_CLOUD_LOCATION=us-central1" >> .env
4. Tworzenie lokalnego serwera MCP
Zanim zaczniesz koordynować pracę agenta walutowego, musisz najpierw utworzyć serwer MCP, aby udostępnić narzędzia, których będzie on potrzebować.
Serwer MCP umożliwia pisanie prostych programów, które udostępniają określone funkcje (np. pobieranie kursów wymiany walut) jako narzędzia. Agent lub nawet kilku agentów może następnie uzyskać dostęp do tych narzędzi za pomocą standardowego protokołu Model Context Protocol (MCP).
Pakiet Pythona FastMCP umożliwia utworzenie serwera MCP, który udostępnia jedno narzędzie o nazwie get_exchange_rate. Narzędzie get_exchange_rate wywołuje przez internet interfejs Frankfurter API, aby uzyskać aktualny kurs wymiany między dwiema walutami.
Kod serwera MCP znajdziesz w pliku 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),
)
)
Aby uruchomić serwer MCP lokalnie, otwórz terminal i uruchom to polecenie (serwer zostanie uruchomiony na porcie http://localhost:8080):
uv run mcp-server/server.py
Sprawdź, czy serwer MCP działa prawidłowo i czy narzędzie get_exchange_rate jest dostępne za pomocą protokołu kontekstu modelu.
W nowym oknie terminala (aby nie zatrzymać lokalnego serwera MCP) uruchom to polecenie:
uv run mcp-server/test_server.py
Powinien się wyświetlić aktualny kurs wymiany 1 USD (dolar amerykański) na EUR (euro):
--- 🛠️ 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
}
} ---
Świetnie! Masz działający serwer MCP z narzędziem, do którego Twój agent będzie mieć dostęp.
Zanim przejdziesz do następnej stacji, zatrzymaj lokalnie działający serwer MCP, wpisując Ctrl+C (lub Command+C na Macu) w terminalu, w którym został uruchomiony.
5. Wdrażanie serwera MCP w Cloud Run
Możesz teraz wdrożyć serwer MCP jako zdalny serwer MCP w Cloud Run 🚀☁️
Zalety zdalnego uruchamiania serwera MCP
Uruchamianie serwera MCP zdalnie w Cloud Run może przynieść kilka korzyści:
- 📈Skalowalność: Cloud Run został zaprojektowany tak, aby szybko skalować się w górę i obsługiwać wszystkie żądania przychodzące. Cloud Run automatycznie skaluje serwer MCP w zależności od zapotrzebowania.
- 👥 Serwer centralny: możesz udostępniać dostęp do centralnego serwera MCP członkom zespołu za pomocą uprawnień IAM, co pozwoli im łączyć się z nim z lokalnych komputerów zamiast uruchamiać własne serwery lokalnie. Jeśli na serwerze MCP zostanie wprowadzona zmiana, skorzystają z niej wszyscy członkowie zespołu.
- 🔐Bezpieczeństwo: Cloud Run umożliwia łatwe wymuszanie uwierzytelnionych żądań. Umożliwia to tylko bezpieczne połączenia z serwerem MCP, co zapobiega nieautoryzowanemu dostępowi.
Przejdź do katalogu mcp-server:
cd mcp-server
Wdróż serwer MCP w Cloud Run:
gcloud run deploy mcp-server --no-allow-unauthenticated --region=us-central1 --source .
Jeśli usługa została wdrożona, zobaczysz komunikat podobny do tego:
Service [mcp-server] revision [mcp-server-12345-abc] has been deployed and is serving 100 percent of traffic.
Uwierzytelnianie klientów MCP
Ponieważ w przypadku opcji --no-allow-unauthenticated wymagane jest uwierzytelnianie, każdy klient MCP łączący się ze zdalnym serwerem MCP będzie musiał się uwierzytelnić.
Więcej informacji na ten temat znajdziesz w oficjalnej dokumentacji Host MCP servers on Cloud Run (Serwery MCP hostowane w Cloud Run), w zależności od tego, gdzie uruchamiasz klienta MCP.
Aby utworzyć uwierzytelniony tunel do zdalnego serwera MCP na komputerze lokalnym, musisz uruchomić serwer proxy Cloud Run.
Domyślnie adres URL usług Cloud Run wymaga, aby wszystkie żądania były autoryzowane za pomocą roli IAM Wywołujący Cloud Run (roles/run.invoker). To powiązanie zasad uprawnień zapewnia, że do uwierzytelniania lokalnego klienta MCP jest używany silny mechanizm zabezpieczeń.
Upewnij się, że Ty i wszyscy członkowie zespołu, którzy próbują uzyskać dostęp do zdalnego serwera MCP, macie rolę uprawnień roles/run.invoker powiązaną z podmiotem zabezpieczeń (kontem Google Cloud).
gcloud run services proxy mcp-server --region=us-central1
Powinny się wyświetlić te dane wyjściowe:
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
Cały ruch do http://127.0.0.1:8080 będzie teraz uwierzytelniany i przekazywany do zdalnego serwera MCP.
Testowanie zdalnego serwera MCP
W nowym terminalu wróć do folderu głównego i ponownie uruchom plik mcp-server/test_server.py, aby sprawdzić, czy zdalny serwer MCP działa.
cd ..
uv run mcp-server/test_server.py
Powinny pojawić się wyniki podobne do tych, które zostały wyświetlone podczas uruchamiania serwera lokalnie:
--- 🛠️ 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
}
} ---
Możesz wysłać zapytanie do logów wdrożonego serwera MCP Cloud Run, aby sprawdzić, czy serwer zdalny został wywołany:
gcloud run services logs read mcp-server --region us-central1 --limit 5
W logach powinny się pojawić te dane wyjściowe:
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}}
Teraz, gdy masz zdalny serwer MCP, możesz przejść do tworzenia agenta. 🤖
6. Tworzenie agenta za pomocą pakietu ADK
Masz już wdrożony serwer MCP, więc teraz możesz utworzyć agenta walutowego za pomocą Agent Development Kit (ADK).
Niedawno udostępniliśmy stabilną wersję 1.0.0 pakietu Agent Development Kit. Ten kamień milowy oznacza, że pakiet ADK w Pythonie jest już gotowy do użycia w środowisku produkcyjnym i stanowi niezawodną i solidną platformę, na której deweloperzy mogą bez obaw tworzyć i wdrażać swoje agenty w środowiskach produkcyjnych.
Pakiet ADK sprawia, że tworzenie agentów jest niezwykle proste, a dzięki wbudowanej obsłudze narzędzi MCP mogą oni łatwo łączyć się z serwerami MCP. Agent walutowy będzie uzyskiwać dostęp do narzędzia get_exchange_rate za pomocą klasy MCPToolset w ADK.
Kod agenta walutowego znajduje się w 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()
Aby szybko przetestować agenta walutowego, możesz skorzystać z interfejsu programistycznego ADK, do którego dostęp uzyskasz, uruchamiając to polecenie: adk web
uv run adk web
Otwórz http://localhost:8000 w przeglądarce, aby zobaczyć i przetestować agenta.
Sprawdź, czy w lewym górnym rogu interfejsu internetowego jako agent jest wybrany currency_agent.
W obszarze czatu zadaj agentowi pytanie, np. „Ile to jest 250 CAD w USD?”. Przed udzieleniem odpowiedzi powinien zadzwonić do naszego get_exchange_ratenarzędzia MCP.
Agent działa. Może odpowiadać na pytania dotyczące przeliczania walut 💸.
7. Protokół Agent2Agent (A2A)
Protokół Agent2Agent (A2A) to otwarty standard zaprojektowany z myślą o umożliwieniu płynnej komunikacji i współpracy między agentami AI. Umożliwia to agentom utworzonym przy użyciu różnych platform i przez różnych dostawców komunikowanie się ze sobą w wspólnym języku, co eliminuje silosy i wspiera interoperacyjność.
A2A umożliwia agentom:
- Odkrywanie: znajdź innych agentów i poznaj ich umiejętności (AgentSkill) oraz możliwości (AgentCapabilities) za pomocą standardowych kart agenta.
- Komunikacja: bezpieczna wymiana wiadomości i danych.
- Współpraca: deleguj zadania i koordynuj działania, aby osiągać złożone cele.
Protokół A2A ułatwia tę komunikację dzięki mechanizmom takim jak „karty agenta”, które działają jak cyfrowe wizytówki, których agenci mogą używać do reklamowania swoich możliwości i informacji o połączeniu.
Teraz musisz udostępnić agenta walutowego za pomocą A2A, aby mógł być wywoływany przez innych agentów i klientów.
A2A Python SDK
Pakiet A2A Python SDK zawiera modele Pydantic dla każdego z wymienionych wyżej zasobów: AgentSkill, AgentCapabilities i AgentCard. Zapewnia interfejs, który przyspiesza tworzenie i integrację z protokołem A2A.
AgentSkill to sposób, w jaki możesz poinformować innych agentów, że agent walutowy ma narzędzie do 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?'],
)
W ramach AgentCard wyświetli umiejętności i możliwości agenta wraz z dodatkowymi informacjami, takimi jak tryby wejścia i wyjścia, które agent może obsługiwać:
# 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],
)
Interfejs AgentExecutor obsługuje podstawową logikę przetwarzania żądań i generowania odpowiedzi/zdarzeń przez agenta A2A. Pakiet A2A Python SDK udostępnia abstrakcyjną klasę bazową a2a.server.agent_execution.AgentExecutor, którą musisz wdrożyć.
Nadszedł czas, aby połączyć to wszystko z agentem walutowym i pokazać moc A2A.
8. Serwer Currency Agent A2A
Teraz przyjrzysz się niektórym fragmentom kodu i zobaczysz, jak łączą się ze sobą różne elementy serwera A2A.
W pliku currency_agent/agent_executor.py znajduje się klasa ADKAgentExecutor, która dziedziczy po abstrakcyjnej klasie A2A AgentExecutor. Obsługuje wywoływanie agenta ADK przez wywoływanie narzędzia ADK Runner, przetwarzanie żądań do agenta oraz konwertowanie danych między formatem google.genai.types używanym przez ADK a formatem a2a.types używanym przez 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
W sekcji currency_agent/__main__.py zainicjujesz AgentSkill i AgentCard oraz utworzysz agenta walutowego ADK. Możesz tam też skonfigurować i uruchomić serwer A2A.
Pakiet SDK A2A w Pythonie udostępnia klasę A2AFastAPIApplication, która upraszcza uruchamianie serwera HTTP zgodnego z A2A. Korzysta z FastAPI jako platformy internetowej i jest zwykle uruchamiana na serwerze ASGI, takim jak 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
Aby uruchomić serwer A2A, w nowym terminalu uruchom to polecenie:
uv run currency_agent/
Jeśli serwer zostanie uruchomiony, dane wyjściowe będą wyglądać tak, jak poniżej, co oznacza, że działa on na porcie 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)
Agent walutowy działa teraz jako serwer A2A i może być wywoływany przez innych agentów lub klientów za pomocą protokołu A2A.
Testowanie serwera A2A
Możesz teraz przetestować serwer, wysyłając do niego żądania za pomocą A2A.
Pakiet A2A Python SDK udostępnia klasę a2a.client.A2AClient, która ułatwia to zadanie.
Plik currency_agent/test_client.py zawiera kod, który uruchamia kilka różnych przypadków testowych na serwerze 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.')
Uruchom testy za pomocą tego polecenia:
uv run currency_agent/test_client.py
Pomyślne przeprowadzenie testu spowoduje:
--- 🔄 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. ---
Udało się Udało Ci się przetestować komunikację z agentem walutowym za pomocą serwera A2A. 🎉
Aby poznać bardziej zaawansowane przypadki użycia, zajrzyj do repozytorium a2a-samples w GitHubie.
Chcesz wdrożyć agenta? Vertex AI Agent Engine zapewnia zarządzane środowisko do wdrażania agentów AI w środowisku produkcyjnym.
9. Gratulacje
Gratulacje! Udało Ci się utworzyć i wdrożyć zdalny serwer MCP, utworzyć agenta walutowego za pomocą pakietu Agent Development Kit (ADK), który łączy się z narzędziami za pomocą MCP, i udostępnić agenta za pomocą protokołu Agent2Agent (A2A). Agent walutowy może teraz wchodzić w interakcje z innymi agentami z dowolnej platformy za pomocą A2A.
Tutaj znajdziesz link do pełnej dokumentacji kodu.
Omówione zagadnienia
- Jak utworzyć lokalny serwer MCP
- Wdrażanie serwera MCP w Cloud Run
- Jak utworzyć agenta za pomocą pakietu ADK, który korzysta z narzędzi MCP
- Udostępnianie agenta ADK jako serwera A2A
- Testowanie serwera A2A za pomocą klienta A2A
Czyszczenie danych
Aby uniknąć obciążenia konta Google Cloud opłatami za zasoby zużyte w tym laboratorium, wykonaj te czynności:
- W konsoli Google Cloud otwórz stronę Zarządzanie zasobami.
- Na liście projektów wybierz projekt, który chcesz usunąć, a potem kliknij Usuń.
- W oknie wpisz identyfikator projektu i kliknij Wyłącz, aby usunąć projekt.