1. Übersicht
KI-Agents werden immer beliebter. Sie revolutionieren die Aufgabenautomatisierung und Entscheidungsfindung, da sie autonom agieren, lernen und mit ihrer Umgebung interagieren können, um Ziele zu erreichen.
Aber wie genau geht man vor, um einen Agent zu erstellen? In diesem Codelab erfahren Sie, wie Sie einen Währungsagenten erstellen, der Währungen verschiedener Länder umrechnen kann. Wir möchten Ihnen die neuesten Technologien vorstellen, damit Sie die Abkürzungen, die Sie vielleicht schon im Internet gesehen haben (MCP, ADK, A2A), besser verstehen.
Model Context Protocol (MCP)
Das Model Context Protocol (MCP) ist ein offenes Protokoll, das standardisiert, wie Anwendungen LLMs Kontext bereitstellen. Das MCP bietet eine standardisierte Möglichkeit, KI-Modelle mit Ressourcen, Prompts und Tools zu verbinden.
Agent Development Kit (ADK)
Das Agent Development Kit (ADK) ist ein flexibles Orchestrierungs-Framework zum Entwickeln und Bereitstellen von KI-Agenten. Das ADK ist modell- und bereitstellungsunabhängig und wurde für die Kompatibilität mit anderen Frameworks entwickelt. Das ADK wurde entwickelt, um die Entwicklung von Agenten an die Softwareentwicklung anzunähern und Entwicklern die Erstellung, Bereitstellung und Orchestrierung von Agentenarchitekturen zu erleichtern, die von einfachen Aufgaben bis hin zu komplexen Workflows reichen.
Agent2Agent-Protokoll (A2A)
Das Agent2Agent-Protokoll (A2A) ist ein offener Standard, der die nahtlose Kommunikation und Zusammenarbeit zwischen KI-Agents ermöglicht. Genau wie das MCP eine standardisierte Möglichkeit bietet, LLMs Zugriff auf Daten und Tools zu gewähren, bietet A2A eine standardisierte Möglichkeit für Agenten, mit anderen Agenten zu kommunizieren. In einer Welt, in der Agenten mit verschiedenen Frameworks und von verschiedenen Anbietern entwickelt werden, bietet A2A eine gemeinsame Sprache, die Silos aufbricht und die Interoperabilität fördert.
Lerninhalte
- Lokalen MCP-Server erstellen
- MCP-Server in Cloud Run bereitstellen
- Agenten mit dem Agent Development Kit erstellen, die MCP-Tools verwenden
- ADK-Agent als A2A-Server bereitstellen
- A2A-Server mit dem A2A-Client testen
Voraussetzungen
2. Hinweis
Projekt erstellen
- Wählen Sie in der Google Cloud Console auf der Seite zur Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.
- Die Abrechnung für das Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.
- Aktivieren Sie Cloud Shell, indem Sie auf diesen Link klicken. Sie können zwischen dem Cloud Shell-Terminal (zum Ausführen von Cloud-Befehlen) und dem Editor (zum Erstellen von Projekten) wechseln, indem Sie in Cloud Shell auf die entsprechende Schaltfläche klicken.
- Wenn Sie mit Cloud Shell verbunden sind, können Sie mit dem folgenden Befehl prüfen, ob Sie bereits authentifiziert sind und das Projekt auf Ihre Projekt-ID festgelegt ist:
gcloud auth list
- Führen Sie den folgenden Befehl in Cloud Shell aus, um zu bestätigen, dass der gcloud-Befehl Ihr Projekt kennt.
gcloud config list project
- Verwenden Sie den folgenden Befehl, um Ihr Projekt festzulegen:
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
- Aktivieren Sie die erforderlichen APIs mit dem folgenden Befehl. Dies kann einige Minuten dauern.
gcloud services enable cloudresourcemanager.googleapis.com \
servicenetworking.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
aiplatform.googleapis.com \
compute.googleapis.com
- Python 3.10 oder höher muss installiert sein.
Informationen zu gcloud-Befehlen und deren Verwendung finden Sie in der Dokumentation.
3. Installation
- Klonen Sie das Repository:
git clone https://github.com/jackwotherspoon/currency-agent.git
cd currency-agent
- Installieren Sie uv (wird zum Verwalten von Abhängigkeiten verwendet):
# 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"
- Umgebungsvariablen konfigurieren (über die Datei
.env
):
Erstellen Sie mit diesem Befehl eine .env
-Datei:
echo "GOOGLE_GENAI_USE_VERTEXAI=TRUE" >> .env \
&& echo "GOOGLE_CLOUD_PROJECT=$PROJECT_ID" >> .env \
&& echo "GOOGLE_CLOUD_LOCATION=us-central1" >> .env
4. Lokalen MCP-Server erstellen
Bevor Sie Ihren Währungs-Agent orchestrieren, erstellen Sie zuerst einen MCP-Server, um die Tools verfügbar zu machen, die Ihr Agent benötigt.
Mit einem MCP-Server können Sie einfache Programme schreiben, um bestimmte Funktionen (z. B. das Abrufen von Währungskursen) als Tools verfügbar zu machen. Ein oder mehrere Agents können dann über das standardisierte Model Context Protocol (MCP) auf diese Tools zugreifen.
Mit dem Python-Paket FastMCP kann ein MCP-Server erstellt werden, der ein einzelnes Tool namens get_exchange_rate
bereitstellt. Das Tool get_exchange_rate
ruft über das Internet die Frankfurter API auf, um den aktuellen Wechselkurs zwischen zwei Währungen abzurufen.
Der Code für den MCP-Server befindet sich in der Datei 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),
)
)
Wenn Sie den MCP-Server lokal starten möchten, öffnen Sie ein Terminal und führen Sie den folgenden Befehl aus (der Server wird auf http://localhost:8080
gestartet):
uv run mcp-server/server.py
Testen Sie, ob der MCP-Server ordnungsgemäß funktioniert und ob das get_exchange_rate
-Tool über das Model Context Protocol zugänglich ist.
Führen Sie in einem neuen Terminal-Fenster (damit der lokale MCP-Server nicht beendet wird) Folgendes aus:
uv run mcp-server/test_server.py
Es sollte der aktuelle Wechselkurs von 1 $ (US-Dollar) zu EUR (Euro) ausgegeben werden:
--- 🛠️ 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
}
} ---
Sehr gut! Sie haben einen funktionierenden MCP-Server mit einem Tool, auf das Ihr Agent zugreifen kann.
Bevor Sie mit der nächsten Station fortfahren, beenden Sie den lokal ausgeführten MCP-Server, indem Sie im Terminal, in dem Sie ihn gestartet haben, Ctrl+C
(oder Command+C
auf dem Mac) ausführen.
5. MCP-Server in Cloud Run bereitstellen
Jetzt können Sie den MCP-Server als Remote-MCP-Server in Cloud Run bereitstellen. 🚀☁️
Vorteile des Remote-Betriebs eines MCP-Servers
Das Remote-Ausführen eines MCP-Servers in Cloud Run bietet mehrere Vorteile:
- 📈 Skalierbarkeit: Cloud Run wurde entwickelt, um schnell horizontal skaliert zu werden, um alle eingehenden Anfragen zu bearbeiten. Cloud Run skaliert Ihren MCP-Server automatisch je nach Bedarf.
- 👥 Zentralisierter Server: Sie können den Zugriff auf einen zentralisierten MCP-Server über IAM-Berechtigungen mit Teammitgliedern teilen. So können sie von ihren lokalen Computern aus eine Verbindung herstellen, anstatt alle ihre eigenen Server lokal auszuführen. Wenn eine Änderung am MCP-Server vorgenommen wird, profitieren alle Teammitglieder davon.
- 🔐 Sicherheit: Cloud Run bietet eine einfache Möglichkeit, authentifizierte Anfragen zu erzwingen. Dadurch sind nur sichere Verbindungen zu Ihrem MCP-Server möglich, was unbefugten Zugriff verhindert.
Wechseln Sie in das Verzeichnis mcp-server
:
cd mcp-server
Stellen Sie den MCP-Server in Cloud Run bereit:
gcloud run deploy mcp-server --no-allow-unauthenticated --region=us-central1 --source .
Wenn Ihr Dienst erfolgreich bereitgestellt wurde, wird eine Meldung wie die folgende angezeigt:
Service [mcp-server] revision [mcp-server-12345-abc] has been deployed and is serving 100 percent of traffic.
MCP-Clients authentifizieren
Da Sie --no-allow-unauthenticated
angegeben haben, um eine Authentifizierung zu erzwingen, muss sich jeder MCP-Client, der eine Verbindung zum Remote-MCP-Server herstellt, authentifizieren.
In der offiziellen Dokumentation unter MCP-Server in Cloud Run hosten finden Sie weitere Informationen zu diesem Thema, je nachdem, wo Sie Ihren MCP-Client ausführen.
Sie müssen den Cloud Run-Proxy ausführen, um einen authentifizierten Tunnel zum Remote-MCP-Server auf Ihrem lokalen Computer zu erstellen.
Standardmäßig erfordert die URL von Cloud Run-Diensten, dass alle Anfragen mit der IAM-Rolle Cloud Run Invoker (roles/run.invoker
) autorisiert werden. Diese IAM-Richtlinienbindung sorgt dafür, dass ein starker Sicherheitsmechanismus zur Authentifizierung Ihres lokalen MCP-Clients verwendet wird.
Sie und alle Teammitglieder, die auf den Remote-MCP-Server zugreifen möchten, müssen die roles/run.invoker
-IAM-Rolle haben, die an ihr IAM-Hauptkonto (Google Cloud-Konto) gebunden ist.
gcloud run services proxy mcp-server --region=us-central1
Es sollte folgende Ausgabe angezeigt werden:
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
Der gesamte Traffic zu http://127.0.0.1:8080
wird jetzt authentifiziert und an den Remote-MCP-Server weitergeleitet.
Remote-MCP-Server testen
Kehren Sie in einem neuen Terminal zum Stammordner zurück und führen Sie die Datei mcp-server/test_server.py
noch einmal aus, um sicherzugehen, dass der Remote-MCP-Server funktioniert.
cd ..
uv run mcp-server/test_server.py
Die Ausgabe sollte in etwa so aussehen wie beim lokalen Ausführen des Servers:
--- 🛠️ 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
}
} ---
Sie können die Logs des bereitgestellten Cloud Run-MCP-Servers abfragen, um zu prüfen, ob der Remote-Server tatsächlich aufgerufen wurde:
gcloud run services logs read mcp-server --region us-central1 --limit 5
In den Logs sollte Folgendes ausgegeben werden:
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}}
Nachdem Sie einen Remote-MCP-Server haben, können Sie mit dem Erstellen eines Agenten fortfahren. 🤖
6. Agent mit dem Agent Development Kit (ADK) erstellen
Sie haben einen MCP-Server bereitgestellt. Jetzt ist es an der Zeit, den Währungsagenten mit dem Agent Development Kit (ADK) zu erstellen.
Das Agent Development Kit wurde vor Kurzem in der stabilen Version 1.0.0 veröffentlicht. Dieser Meilenstein bedeutet, dass das Python ADK jetzt produktionsreif ist und Entwicklern eine zuverlässige und robuste Plattform bietet, auf der sie ihre Agents in Live-Umgebungen erstellen und bereitstellen können.
Mit dem ADK lassen sich KI-Agenten sehr einfach erstellen und problemlos mit MCP-Servern verbinden. MCP-Tools werden dabei automatisch unterstützt. Der Währungs-Agent greift mit der MCPToolset-Klasse des ADK auf das get_exchange_rate
-Tool zu.
Der Code für den Währungs-Agent befindet sich in 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()
Um den Währungs-Agenten schnell zu testen, können Sie die Entwickler-UI des ADK verwenden. Rufen Sie sie mit dem folgenden Befehl auf: adk web
uv run adk web
Rufen Sie in einem Browser http://localhost:8000
auf, um den Agenten zu sehen und zu testen.
Achten Sie darauf, dass currency_agent
oben links in der Web-UI als Agent ausgewählt ist.
Fragen Sie Ihren Agent im Chatbereich z. B. „What is 250 CAD to USD?“ (Wie viel sind 250 CAD in USD?). Der Kundenservicemitarbeiter sollte das MCP-Tool get_exchange_rate
aufrufen, bevor er eine Antwort gibt.
Der Agent funktioniert. Es kann Anfragen zur Währungsumrechnung beantworten. 💸
7. Agent2Agent-Protokoll (A2A)
Das Agent2Agent-Protokoll (A2A) ist ein offener Standard, der die nahtlose Kommunikation und Zusammenarbeit zwischen KI-Agents ermöglicht. So können Agents, die mit verschiedenen Frameworks und von verschiedenen Anbietern erstellt wurden, in einer gemeinsamen Sprache miteinander kommunizieren. Das trägt dazu bei, Silos aufzubrechen und die Interoperabilität zu fördern.
Mit A2A können Agents:
- Entdecken:Mit standardisierten Agent-Karten können Sie andere Agents und ihre Fähigkeiten (AgentSkill) und Funktionen (AgentCapabilities) finden.
- Kommunizieren:Nachrichten und Daten sicher austauschen.
- Zusammenarbeiten:Aufgaben delegieren und Aktionen koordinieren, um komplexe Ziele zu erreichen.
Das A2A-Protokoll erleichtert diese Kommunikation durch Mechanismen wie „Agent Cards“, die als digitale Visitenkarten dienen, mit denen Agenten ihre Funktionen und Verbindungsinformationen bewerben können.
Jetzt ist es an der Zeit, den Währungs-Agent über A2A verfügbar zu machen, damit er von anderen Agents und Clients aufgerufen werden kann.
A2A Python SDK
Das A2A Python SDK bietet Pydantic-Modelle für jede der oben genannten Ressourcen: AgentSkill, AgentCapabilities und AgentCard. Dies bietet eine Schnittstelle, um die Entwicklung und Integration mit dem A2A-Protokoll zu beschleunigen.
Mit einem AgentSkill
werben Sie bei anderen Kundenservicemitarbeitern dafür, dass der Währungskundenservicemitarbeiter ein Tool für get_exchange_rate
hat:
# 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?'],
)
Im Rahmen von AgentCard
werden dann die Fähigkeiten und Funktionen des Agents zusammen mit zusätzlichen Details wie den Eingabe- und Ausgabemodi aufgeführt, die der Agent verarbeiten kann:
# 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],
)
Die AgentExecutor-Schnittstelle enthält die Kernlogik dafür, wie ein A2A-Agent Anfragen verarbeitet und Antworten/Ereignisse generiert. Das A2A Python SDK bietet eine abstrakte Basisklasse a2a.server.agent_execution.AgentExecutor
, die Sie implementieren müssen.
Es ist an der Zeit, alles mit dem Währungs-Agenten zu kombinieren und die Leistungsfähigkeit von A2A zu demonstrieren.
8. Currency Agent A2A Server
Sehen Sie sich nun einige Codeabschnitte an und erfahren Sie, wie die verschiedenen Teile eines A2A-Servers zusammenarbeiten.
In der Datei currency_agent/agent_executor.py
befindet sich die Klasse ADKAgentExecutor
, die von der abstrakten A2A-Klasse AgentExecutor
abgeleitet wird. Es ruft den ADK-Agenten auf, indem es den ADK-Runner aufruft, Anfragen an den Agenten verarbeitet und zwischen google.genai.types
, das vom ADK verwendet wird, und a2a.types
, das von A2A verwendet wird, konvertiert.
# ... 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
In currency_agent/__main__.py
initialisieren Sie den AgentSkill und die AgentCard und erstellen den ADK-Währungsagenten. Hier richten Sie auch den A2A-Server ein und starten ihn.
Das A2A Python SDK bietet eine A2AFastAPIApplication
-Klasse, die das Ausführen eines A2A-kompatiblen HTTP-Servers vereinfacht. Es verwendet FastAPI für das Webframework und wird in der Regel mit einem ASGI-Server wie Uvicorn ausgeführt.
# ... 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
Führen Sie den folgenden Befehl in einem neuen Terminal aus, um den A2A-Server auszuführen:
uv run currency_agent/
Wenn der Server erfolgreich gestartet wird, sieht die Ausgabe so aus. Sie gibt an, dass der Server auf Port 10000 ausgeführt wird:
[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)
Der Währungs-Agent wird jetzt erfolgreich als A2A-Server ausgeführt und kann von anderen Agents oder Clients über das A2A-Protokoll aufgerufen werden.
A2A-Server testen
Sie können den Server jetzt testen, indem Sie ihm einige Anfragen über A2A senden.
Das A2A Python SDK bietet eine a2a.client.A2AClient
-Klasse, die dies für Sie vereinfacht.
Die Datei currency_agent/test_client.py
enthält Code, der verschiedene Testläufe für den A2A-Server ausführt.
# ... 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.')
Führen Sie die Tests mit dem folgenden Befehl aus:
uv run currency_agent/test_client.py
Ein erfolgreicher Testlauf führt zu Folgendem:
--- 🔄 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. ---
Es funktioniert. Sie haben erfolgreich getestet, dass Sie über einen A2A-Server mit dem Währungs-Agent kommunizieren können. 🎉
Im Repository „a2a-samples“ auf GitHub finden Sie weitere erweiterte Anwendungsfälle.
Möchten Sie Ihren Agent bereitstellen? Vertex AI Agent Engine bietet eine verwaltete Umgebung für die Bereitstellung von KI-Agents in der Produktion.
9. Glückwunsch
Glückwunsch! Sie haben erfolgreich einen Remote-MCP-Server erstellt und bereitgestellt, einen Währungs-Agenten mit dem Agent Development Kit (ADK) erstellt, der über MCP eine Verbindung zu Tools herstellt, und Ihren Agenten über das Agent2Agent-Protokoll (A2A) bereitgestellt. Der Währungs-Agent kann jetzt über A2A mit anderen Agents beliebiger Frameworks interagieren.
Hier finden Sie einen Link zur vollständigen Codedokumentation.
Behandelte Themen
- Lokalen MCP-Server erstellen
- MCP-Server in Cloud Run bereitstellen
- Agenten mit dem Agent Development Kit erstellen, die MCP-Tools verwenden
- ADK-Agent als A2A-Server bereitstellen
- A2A-Server mit dem A2A-Client testen
Bereinigen
So vermeiden Sie, dass Ihrem Google Cloud-Konto die in diesem Lab verwendeten Ressourcen in Rechnung gestellt werden:
- Wechseln Sie in der Google Cloud Console zur Seite Ressourcen verwalten.
- Wählen Sie in der Projektliste das Projekt aus, das Sie löschen möchten, und klicken Sie auf Löschen.
- Geben Sie im Dialogfeld die Projekt-ID ein und klicken Sie auf Beenden, um das Projekt zu löschen.