Dynamiczne wykrywanie serwerów MCP za pomocą rejestru agentów i ADK

1. Wprowadzenie

Z tego ćwiczenia dowiesz się, jak używać rejestru agentów Google Cloud i pakietu Agent Development Kit (ADK) do dynamicznego wykrywania i wywoływania zarejestrowanych serwerów Model Context Protocol (MCP) w czasie działania, co eliminuje konieczność kodowania na stałe punktów końcowych i konfiguracji narzędzi.

W tym module do utworzenia aplikacji używamy pakietu ADK i języka programowania Python. Te same zasady można stosować w innych językach programowania i przy użyciu innych platform do tworzenia agentów.

Jakie zadania wykonasz

  • Włączanie i konfigurowanie Google Cloud Agent Registry w projekcie
  • Wdrażanie serwera MCP w Cloud Run i rejestrowanie go w rejestrze agentów
  • Tworzenie agenta ADK, który wywołuje serwer MCP za pomocą zakodowanego na stałe adresu URL
  • Zmień agenta, aby wykrywał serwer MCP w rejestrze agentów na podstawie jego tożsamości
  • Zmienianie agenta w celu wykrywania serwerów MCP w czasie działania za pomocą wyszukiwania semantycznego

Czego potrzebujesz

  • nowoczesna przeglądarka (np. Chrome),
  • projekt Google Cloud z włączonymi rozliczeniami,
  • konto z uprawnieniami do tworzenia zasobów i zarządzania nimi oraz konfigurowania zasad uprawnień w projekcie Google Cloud;

2. Konfigurowanie środowiska

Instrukcje w tym ćwiczeniu z programowania zakładają, że uruchamiasz polecenia w Cloud Shell w konsoli Google Cloud. Cloud Shell to środowisko wiersza poleceń działające w Google Cloud, które zawiera niezbędne narzędzia. Jeśli w środowisku lokalnym masz zainstalowany interfejs wiersza poleceń gcloud, git i Pythona 3, możesz też uruchamiać polecenia w tym środowisku.

Utwórz projekt Google Cloud

  1. W Google Cloud Console na stronie selektora projektów utwórz projekt Google Cloud.
  2. Sprawdź, czy w projekcie Cloud włączone są płatności. Dowiedz się, jak sprawdzić, czy w projekcie są włączone płatności.

Uruchamianie terminala Cloud Shell

  1. Za pomocą selektora zasobów wybierz utworzony projekt w chmurze Google.
  2. Kliknij Aktywuj Cloud Shell Ikona aktywowania Cloud Shell na górze konsoli Google Cloud.
  3. Po połączeniu z Cloud Shell sprawdź uwierzytelnianie:
    gcloud auth list
  4. Potwierdź, że projekt jest skonfigurowany:
    gcloud config get project
  5. Jeśli projekt nie jest ustawiony zgodnie z oczekiwaniami lub jeśli zamiast Cloud Shell używasz środowiska na komputerze lokalnym, skonfiguruj identyfikator projektu:
    export GOOGLE_CLOUD_PROJECT=<YOUR_PROJECT_ID>
gcloud config set project "${GOOGLE_CLOUD_PROJECT}"

Wybieranie projektu w chmurze Google Cloud i ustawianie zmiennych środowiskowych

  1. Włącz wymagane Google Cloud APIs:
    gcloud services enable \
  agentregistry.googleapis.com \
  aiplatform.googleapis.com \
  cloudbuild.googleapis.com \
  run.googleapis.com \
  --project="${GOOGLE_CLOUD_PROJECT}"

Konfigurowanie środowiska Pythona

  1. Utwórz i aktywuj środowisko wirtualne oraz zainstaluj wymagane zależności:
    mkdir -p adk-dynamic-discovery && cd adk-dynamic-discovery
python3 -m venv .venv
source .venv/bin/activate
pip install "fastmcp<5.0.0" "google-adk[mcp,a2a,agent-identity]>=2.5.0,<3.0.0"

3. Tworzenie i rejestrowanie serwera Zoo MCP

W tym kroku utworzysz i zarejestrujesz serwer MCP Zoo. Serwer udostępnia 3 narzędzia: list_species – wyświetla listę wszystkich gatunków żyjących w zoo, get_animals_by_species – wyświetla listę zwierząt danego gatunku, oraz get_animal_details – wyświetla dodatkowe informacje o konkretnym zwierzęciu.

Wszystkie pliki związane z tym krokiem znajdziesz na GitHub.

Utwórz serwer MCP

  1. Konfigurowanie folderu i projektu w Pythonie na potrzeby serwera MCP
    mkdir mcp-on-cloudrun && cd mcp-on-cloudrun
uv init --description "Example of deploying an MCP server on Cloud Run" --bare --python 3.13
uv add "fastmcp<5.0.0" --no-sync
  2. Pobierz kod serwera MCP:
    curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step3/server.py -o server.py

Wdrażanie serwera MCP w Cloud Run i rejestrowanie go w rejestrze agentów

Następnie utworzysz Dockerfile dla serwera i wdrożysz serwer w Cloud Run.

  1. Utwórz Dockerfile:
    cat << EOF > Dockerfile
# Use the official Python image
FROM python:3.13-slim

# Install uv
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

# Install the project into /app
COPY . /app
WORKDIR /app

# Allow statements and log messages to immediately appear in the logs
ENV PYTHONUNBUFFERED=1

# Install dependencies
RUN uv sync

# Run the FastMCP server
CMD ["uv", "run", "server.py"]
EOF
  2. Utwórz wyznaczone konto usługi dla serwera:
    gcloud iam service-accounts create zoo-mcp-server-sa \
    --display-name="Zoo MCP Server Service Account"
  3. wdrożyć serwer w regionie us-west1,
    gcloud run deploy zoo-mcp-server \
    --service-account=zoo-mcp-server-sa@$GOOGLE_CLOUD_PROJECT.iam.gserviceaccount.com \
    --region=us-west1 \
    --source=. \
    --labels=dev-tutorial=codelab-dynamic-mcp-discovery \
    --invoker-iam-check \
    --no-allow-unauthenticated
    Jeśli pojawi się ten komunikat, potwierdź utworzenie nowego repozytorium Artifact Registry. Jeśli w projekcie nie masz repozytorium Artifact Registry dla regionu us-west1, pojawi się ten komunikat:
    Deploying from source requires an Artifact Registry Docker repository to store built containers. A repository named
[cloud-run-source-deploy] in region [us-west1] will be created.

Do you want to continue (Y/n)?
  4. Zarejestruj serwer MCP w rejestrze agentów w lokalizacji „global”:
    fastmcp inspect server.py --format mcp | jq 'del(..|nulls) | del(.tools[]?._meta?)' > toolspec.json
MCP_URL=$(gcloud run services describe zoo-mcp-server --region=us-west1 --format='value(status.url)')
gcloud agent-registry services create zoo-mcp-server \
    --display-name="zoo-mcp-server" \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --location=global \
    --mcp-server-spec-type="tool-spec" \
    --mcp-server-spec-content="toolspec.json" \
    --interfaces="url=${MCP_URL}/mcp,protocolBinding=jsonrpc"
    Te polecenia generują plik toolspec.json z manifestem narzędzia i wywołują interfejs API rejestru agentów, aby zarejestrować serwer.
  5. Sprawdź, czy agent pojawia się w rejestrze:
    gcloud agent-registry mcp-servers list \
    --location=global \
    --filter="display_name=zoo-mcp-server"
    Zobaczysz dane wyjściowe YAML z informacjami o serwerze MCP, takie jak te:
    ---
attributes:
  agentregistry.googleapis.com/system/RuntimeReference:
    uri: //agentregistry.googleapis.com/projects/[PROJECT_NUMBER]/locations/global/services/zoo-mcp-server
createTime: '<create time datetime in ISO 8601>'
displayName: zoo-mcp-server
interfaces:
- protocolBinding: JSONRPC
  url: https://zoo-mcp-server-[PROJECT_NUMBER].us-west1.run.app/mcp
mcpServerId: urn:mcp:projects-[PROJECT_NUMBER]:projects:[PROJECT_NUMBER]:locations:global:agentregistry:services:zoo-mcp-server
name: projects/[PROJECT_ID]/locations/global/mcpServers/agentregistry-00000000-0000-0000-1234-56789abcdef
tools:
- annotations:
    idempotentHint: true
    readOnlyHint: true
  description: |-
    Returns every species in the zoo. Use it first for zoo-wide questions
    such as counting all animals or listing everything the zoo has.
  name: list_species
- annotations:
    idempotentHint: true
    readOnlyHint: true
  description: |-
    Retrieves all animals of a specific species from the zoo.
    Can also be used to collect the base data for aggregate queries
    of animals of a specific species - like counting the number of penguins
    or finding the oldest lion.
  name: get_animals_by_species
- annotations:
    idempotentHint: true
    readOnlyHint: true
  description: Retrieves the details of a specific animal by its name.
  name: get_animal_details
updateTime: '<last updated datetime in ISO 8601>'

4. Tworzenie podstawowego agenta ADK do korzystania z serwera MCP Zoo

W tym kroku utworzysz agenta ADK, który będzie odpowiadać na pytania użytkowników dotyczące zoo, korzystając z zakodowanego na stałe adresu URL serwera MCP Zoo. Kodowanie na stałe wymaga znajomości adresu URL serwera MCP PRZED wdrożeniem agenta.

Pliki źródłowe do tego kroku znajdziesz na GitHub.

Tworzenie aplikacji agenta

Utwórz szkielet projektu ADK i agenta Zoo, który korzysta z serwera MCP Zoo.

  1. Skonfiguruj folder i projekt dla agenta Zoo:
    mkdir ../sample-agent && cd ../sample-agent
uv init --description "Example of ADK agent to showcase dynamic MCP discovery" --bare --python 3.13
uv add "google-adk[mcp,a2a,agent-identity]>=2.5.0,<3.0.0" --no-sync
  2. Utwórz folder pakietu agenta i pobierz plik __init__.py:
    mkdir zoo_agent
curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step4/zoo_agent/__init__.py -o zoo_agent/__init__.py
  3. Utwórz plik .env i wypełnij go zmiennymi środowiskowymi, które są używane w tym i kolejnych krokach:
    MCP_URL=$(gcloud run services describe zoo-mcp-server --region=us-west1 --format='value(status.url)')
cat << EOF > zoo_agent/.env
GOOGLE_GENAI_USE_ENTERPRISE=TRUE
GOOGLE_CLOUD_LOCATION=global
GOOGLE_CLOUD_PROJECT="${GOOGLE_CLOUD_PROJECT}"
GOOGLE_CLOUD_PROJECT_NUMBER=$(gcloud projects describe ${GOOGLE_CLOUD_PROJECT} \
   --format='value(projectNumber)')
MCP_SERVER_URL="${MCP_URL}"
EOF
    To polecenie zapisuje wszystkie zmienne środowiskowe wymagane w tym kroku i kolejnych krokach wykrywania w pliku zoo_agent/.env, dzięki czemu nie musisz później ponownie tworzyć tego pliku. W tym kroku używane są te zmienne:
    • GOOGLE_GENAI_USE_ENTERPRISE, GOOGLE_CLOUD_PROJECT i GOOGLE_CLOUD_LOCATION: używane niejawnie przez ADK i pakiet SDK Google GenAI do kierowania wywołań modelu Gemini przez Vertex AI w Twoim projekcie w chmurze i lokalizacji Google Cloud (global). Przechowuje też lokalizację rejestru agentów, w którym zarejestrowany jest serwer MCP.
    • GOOGLE_CLOUD_PROJECT_NUMBER: nie jest używany w tym kroku; jest wstępnie wypełniony do użycia w następnym kroku.
    • MCP_SERVER_URL: przeczytaj bezpośrednio w zoo_agent/agent.py, aby skonfigurować McpToolset za pomocą zakodowanego na stałe adresu URL Cloud Run serwera MCP Zoo.
  4. Pobierz plik agent.py z kodem agenta Zoo, który łączy się z serwerem MCP Zoo:
    curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step4/zoo_agent/agent.py -o zoo_agent/agent.py

Jak działa autoryzacja serwera MCP

Ponieważ usługa Cloud Run zoo-mcp-server została wdrożona z flagami --invoker-iam-check i --no-allow-unauthenticated, każde żądanie HTTP wysyłane do serwera MCP musi zawierać podpisany przez Google token identyfikacji w nagłówku Authorization. Token musi identyfikować element wywołujący, który ma uprawnienia run.routes.invoke w usłudze.

W pliku zoo_agent/agent.py klasa _GoogleAuthHeaders pobiera dane uwierzytelniające tokena identyfikacji na jeden z dwóch sposobów. Najpierw wywołuje funkcję fetch_id_token_credentials(), aby uzyskać token identyfikacji dla adresu URL usługi z serwera metadanych lub klucza konta usługi. Jeśli to się nie uda, użyje tokena identyfikacji domyślnego uwierzytelniania aplikacji (ADC) użytkownika. Za każdym razem, gdy instancja jest wywoływana, w razie potrzeby odświeża token i zwraca nagłówek Authorization:

class _GoogleAuthHeaders:
    def __init__(self, audience: str):
        self._request = Request()
        self._id_token = False
        try:
            self._credentials = id_token.fetch_id_token_credentials(audience, request=self._request)
        except exceptions.DefaultCredentialsError:
            self._credentials, _ = google.auth.default()
            self._id_token = True
            if not isinstance(self._credentials, credentials.Credentials):
                raise

    def __call__(self, unused_context: ReadonlyContext | None = None) -> dict[str, str]:
        if not self._credentials.valid or (self._id_token and not self._credentials.id_token):
            self._credentials.refresh(self._request)
        token = self._credentials.id_token if self._id_token else self._credentials.token
        return {"Authorization": f"Bearer {token}"}

Agent przekazuje instancję klasy do McpToolset jako argument header_provider. Pakiet ADK wywołuje dostawcę nagłówków i dołącza nagłówki, które zwraca, do każdego żądania wysyłanego do serwera MCP. Ponieważ klient HTTP jest tworzony przez ADK, ten sam kod działa niezależnie od zainstalowanej wersji pakietu MCP Python SDK i jego biblioteki HTTP:

zoo_mcp_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url=f"{MCP_SERVER_URL}/mcp",
        timeout=30.0,
    ),
    header_provider=_GoogleAuthHeaders(MCP_SERVER_URL),
)

Żądany odbiorca to adres URL usługi (MCP_SERVER_URL), a adres URL połączenia zawiera ścieżkę /mcp, pod którą serwer FastMCP obsługuje protokół MCP. Limit czasu wynoszący 30 sekund pozostawia miejsce na uruchomienia „na zimno” Cloud Run.

Testowanie agenta za pomocą środowiska testowego ADK w środowisku lokalnym

Zanim przetestujesz agenta, upewnij się, że Twoje konto użytkownika może wywołać serwer MCP.

  1. Tylko terminal lokalny (pomiń ten krok w Cloud Shell): zainicjuj domyślne uprawnienia aplikacji przy użyciu danych logowania użytkownika:
    gcloud auth application-default login
    Postępuj zgodnie z instrukcjami, aby dokończyć inicjowanie ADC.
  2. (Opcjonalnie) Przyznaj swojemu kontu użytkownika uprawnienia do wywoływania zdalnego serwera MCP:
    gcloud run services add-iam-policy-binding zoo-mcp-server \
   --region=us-west1 \
   --project=${GOOGLE_CLOUD_PROJECT} \
   --member=user:$(gcloud config get-value account) \
   --role='roles/run.invoker'
  3. Uruchom agenta za pomocą polecenia adk run:
    adk run zoo_agent
    lub uruchom interfejs internetowy ADK za pomocą polecenia adk web:
    adk web --port 8080
    a następnie otwórz w przeglądarce adres URL, który zostanie wyświetlony przez polecenie.Jeśli używasz Cloud Shell, dodaj argument --allow_origins, aby umożliwić otwieranie interfejsu internetowego z Cloud Shell:
    adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
    Po uruchomieniu polecenia znajdź ikonę Podgląd w przeglądarceIkona podglądu w przeglądarce na pasku narzędzi Cloud Shell (po prawej stronie). Kliknij ją i wybierz Podgląd na porcie 8080. Cloud Shell otworzy nową kartę przeglądarki z interfejsem internetowym ADK.
  4. Rozpocznij czat z agentem. Możesz na przykład zapytać o zwierzęta w zoo:
    How many animals are in the zoo?
    Agent wywołuje funkcję list_species, a następnie get_animals_by_species dla każdego gatunku, aby obliczyć odpowiedź.

Interfejs internetowy ADK umożliwia sprawdzenie, jak działa agent, w tym jego wywołania serwera MCP Zoo. Aby dowiedzieć się, jak to zrobić, zapoznaj się z dokumentacją interfejsu internetowego.

5. Dynamiczne wykrywanie punktu końcowego serwera MCP

W tym kroku zmienisz agenta, aby wykrywał punkt końcowy serwera MCP Zoo zarejestrowany w Agent Registry, zamiast kodować na stałe jego URL punktu końcowego.

Wszystkie pliki związane z tym krokiem znajdziesz na GitHub.

Aby znaleźć i pobrać adres URL serwera, używasz integracji ADK z rejestrem agentów. Aby go wdrożyć, potrzebujesz tych informacji:

  • Numer projektu i lokalizacja rejestru agentów, w którym zarejestrowany jest serwer MCP (global).
  • Identyfikator usługi użyty do zarejestrowania serwera MCP (zoo-mcp-server).

Aktualizowanie agenta pod kątem korzystania z rejestru agentów

  1. Pobierz z GitHuba zaktualizowany plik agent.py, aby zastąpić pierwszą wersję agenta:
    curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step5/zoo_agent/agent.py -o zoo_agent/agent.py

    Zmienne środowiskowe w tym kroku

    Po tej aktualizacji agent korzysta z tych zmiennych środowiskowych z zoo_agent/.env:
    • GOOGLE_CLOUD_PROJECT_NUMBER: teraz wymagane do utworzenia identyfikatora URN tożsamości serwera MCP (MCP_SERVER_URN).
    • GOOGLE_CLOUD_PROJECT, GOOGLE_GENAI_USE_ENTERPRISE i GOOGLE_CLOUD_LOCATION: używane przez ADK do konfigurowania dostępu do Vertex AI. Lokalizacja jest też używana do wykrywania serwera MCP.
    • Nie jest już wymagana: zmienna MCP_SERVER_URL nie jest już używana, ponieważ agent pobiera adres URL punktu końcowego serwera MCP z Agent Registry, a nie ze środowiska.

    Jak działa tożsamość i wykrywanie serwera MCP w AgentRegistry

    W Agent Registry każdy zarejestrowany serwer MCP ma unikalny identyfikator przechowywany w polu mcpServerId, sformatowany jako jednolity identyfikator zasobu (URN). Możesz go zobaczyć w wyniku polecenia gcloud agent-registry mcp-servers list w kroku 3. Agent rekonstruuje ten identyfikator na podstawie zmiennych środowiskowych:
    PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
PROJECT_NUMBER = os.environ["GOOGLE_CLOUD_PROJECT_NUMBER"]
LOCATION = os.environ.get("GOOGLE_CLOUD_LOCATION", "global")
MCP_SERVER_URN = f"urn:mcp:projects-{PROJECT_NUMBER}:projects:{PROJECT_NUMBER}:locations:{LOCATION}:agentregistry:services:zoo-mcp-server"
    URN koduje numer projektu rejestru (projects-{PROJECT_NUMBER}), numer projektu i lokalizację, w której zarejestrowany jest serwer (projects:{PROJECT_NUMBER}:locations:{LOCATION}), oraz identyfikator usługi użyty w poleceniu gcloud agent-registry services create (agentregistry:services:zoo-mcp-server).Agent wyświetla listę zarejestrowanych serwerów MCP za pomocą registry.list_mcp_servers() i dopasowuje mcpServerId każdego serwera do MCP_SERVER_URN. Następnie przekazuje zasób pasującego serwera name (projects/.../locations/.../mcpServers/...) do registry.get_mcp_toolset(), który odczytuje adres URL punktu końcowego z zarejestrowanego interfejsu serwera (https://...run.app/mcp) i zwraca McpToolset połączony z tym adresem URL:
    header_provider = _GoogleAuthHeaders()
registry = AgentRegistry(
    project_id=PROJECT_ID,
    location=LOCATION,
    header_provider=header_provider,
)

mcp_server_name = next(
    (
        s["name"]
        for s in registry.list_mcp_servers().get("mcpServers", [])
        if s.get("mcpServerId") == MCP_SERVER_URN
    ),
    None,
)
if not mcp_server_name:
    raise RuntimeError(f"MCP server '{MCP_SERVER_URN}' not found in Agent Registry.")
zoo_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)
header_provider.bind(zoo_mcp_toolset.connection_params.url)
zoo_mcp_toolset.connection_params.timeout = 30.0
    get_mcp_toolset() dodaje do nazw narzędzi prefiks z wyświetlaną nazwą serwera, więc agent widzi narzędzia jako zoo_mcp_server_list_species, zoo_mcp_server_get_animals_by_species i zoo_mcp_server_get_animal_details. Prefiks zapobiega kolizjom nazw, gdy agent używa narzędzi z wielu serwerów MCP. Ostatni wiersz zwiększa limit czasu połączenia z domyślnych 5 sekund, aby zapewnić czas na uruchomienie Cloud Run na zimno.

    Jak zmienia się autoryzacja w przypadku AgentRegistry

    AgentRegistry uwierzytelnia własne wywołania interfejsu API, takie jak list_mcp_servers(), za pomocą ADC, więc dostawca nagłówka jest potrzebny tylko do wywołań wykrytego serwera MCP. Ponieważ AgentRegistry tworzy McpToolset wewnętrznie, przekazujesz dostawcę nagłówka do konstruktora AgentRegistry.W poprzednim kroku odbiorcami była wartość MCP_SERVER_URL, którą znasz z góry. Teraz adres URL jest znany dopiero po wykryciu. ADK wywołuje dostawcę nagłówka, gdy agent wysyła żądanie do serwera MCP, a nie podczas tworzenia AgentRegistry. Dzięki temu agent może najpierw utworzyć dostawcę i powiązać go z wykrytym punktem końcowym po zwróceniu get_mcp_toolset(). Metoda bind() pobiera dane logowania w taki sam sposób jak konstruktor w poprzednim kroku, a odbiorcy są pobierani z wykrytego adresu URL punktu końcowego bez ścieżki /mcp:
        def bind(self, endpoint_url: str) -> None:
        parsed = urlparse(endpoint_url)
        audience = f"{parsed.scheme}://{parsed.netloc}"
        try:
            self._credentials = id_token.fetch_id_token_credentials(audience, request=self._request)
        except exceptions.DefaultCredentialsError:
            self._credentials, _ = google.auth.default()
            self._id_token = True
            if not isinstance(self._credentials, credentials.Credentials):
                raise
    Metoda __call__ jest taka sama jak w poprzednim kroku, ale zgłasza błąd, jeśli zostanie wywołana przed bind().
  2. Uruchom agenta, aby przetestować zmianę. Użyj polecenia adk run:
    adk run zoo_agent
    lub uruchom interfejs internetowy ADK za pomocą polecenia adk web:
    adk web --port 8080
    a następnie otwórz w przeglądarce adres URL, który zostanie wyświetlony przez polecenie.Jeśli używasz Cloud Shell, dodaj argument --allow_origins, aby umożliwić otwieranie interfejsu internetowego z Cloud Shell:
    adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
    Zadaj agentowi to samo pytanie co w poprzednim kroku. Odpowiedź jest taka sama, ale w interfejsie internetowym ADK wywołania narzędzi są teraz wyświetlane jako zoo_mcp_server_list_species i zoo_mcp_server_get_animals_by_species.

Dzięki tej zmianie nie trzeba już podawać dokładnego adresu URL serwera MCP. Nadal musisz jednak znać numer projektu, lokalizację rejestru i identyfikator usługi użyty do zarejestrowania serwera MCP, a wykrywanie jest wykonywane statycznie podczas inicjowania agenta.

6. Używanie wyszukiwania semantycznego na serwerze MCP w ramach dynamicznego wykrywania

W tym kroku ulepszysz dynamiczne wykrywanie serwera MCP, zastępując dokładne dopasowanie URN podczas uruchamiania wyszukiwaniem semantycznym w czasie działania. Dzięki temu nie trzeba znać numeru projektu, w którym wdrożono serwer, ani dokładnej tożsamości serwera (MCP_SERVER_URN). Wykrywanie narzędzi jest też zintegrowane bezpośrednio ze standardowym procesem działania agenta, a nie obsługiwane podczas inicjowania.

Wszystkie pliki związane z tym krokiem znajdziesz na GitHub.

Aktualizowanie agenta w celu korzystania z semantycznego wykrywania w czasie działania

  1. Pobierz zaktualizowany plik agent.py z GitHuba, aby zastąpić istniejącą implementację:
    curl -fsSL https://raw.githubusercontent.com/GoogleCloudPlatform/devrel-demos/main/codelabs/agent-registry-dynamic-discovery-with-adk/steps/step6/zoo_agent/agent.py -o zoo_agent/agent.py

    Zmienne środowiskowe w tym kroku

    Ponieważ agent nie tworzy już nazwy URN serwera MCP, wymagana konfiguracja jest jeszcze prostsza:
    • GOOGLE_CLOUD_PROJECT, GOOGLE_GENAI_USE_ENTERPRISE i GOOGLE_CLOUD_LOCATION: używane przez ADK do konfigurowania dostępu do Vertex AI. Lokalizacja jest też używana do wykrywania serwera MCP.
    • Nie jest już wymagane: GOOGLE_CLOUD_PROJECT_NUMBER nie jest już potrzebne (oprócz MCP_SERVER_URL, które zostało już usunięte w poprzednim kroku).
    Zamiast wywoływać registry.list_mcp_servers() podczas inicjowania modułu i porównywać każdy wpis z utworzonym MCP_SERVER_URN, agent.py definiuje teraz funkcję narzędzia do wykrywania discover_zoo_mcp_servers, która wywołuje registry.search_mcp_servers() z search_type="SEMANTIC":
    def discover_zoo_mcp_servers(query: str, tool_context: ToolContext) -> dict[str, Any]:
    """Discovers Zoo MCP server(s) matching the query using semantic search.

    Args:
        query: Semantic description of zoo capabilities needed (e.g. 'zoo animal info')
    """
    cached = _fresh_discovery(tool_context.state)
    if cached:
        return {
            "status": "already_discovered",
            "message": "Zoo MCP servers are already discovered.",
            "servers": cached["servers"],
        }

    try:
        response = registry.search_mcp_servers(
            search_string=query,
            search_type="SEMANTIC",
            page_size=5,
        )
    except Exception as e:
        return {
            "status": "error",
            "message": f"Agent Registry search failed: {e}",
        }

    found_servers = response.get("mcpServers", [])
    if not found_servers:
        return {
            "status": "not_found",
            "message": f"No MCP servers found in Agent Registry at {LOCATION} matching '{query}'.",
        }

    valid_servers = [s["name"] for s in found_servers if "name" in s]
    tool_context.state[DISCOVERY_STATE_KEY] = {
        "servers": valid_servers,
        "discovered_at": time.time(),
    }
    return {
        "status": "discovered",
        "message": f"Successfully discovered {len(valid_servers)} Zoo MCP server(s).",
        "servers": valid_servers,
    }
    Dzięki temu agent może wyszukiwać w rejestrze agentów w czasie działania, używając opisu potrzebnych funkcji w języku naturalnym (np. "zoo animals, exhibits, habitat, care, feeding schedule"), i zapisywać wykryte nazwy zasobów serwera w stanie sesji ADK (tool_context.state).

    Jak dynamiczne wykrywanie jest zintegrowane z przepływem pracy agenta

    Aby podłączyć narzędzia z serwerów wykrytych w czasie działania i przeprowadzić agenta przez proces wykrywania i wywoływania, agent.py łączy niestandardową implementację BaseToolset (RegistryDiscoveredToolset) z dynamicznym dostawcą instrukcji (zoo_guide_instruction):
    1. Dynamiczny zestaw narzędzi (RegistryDiscoveredToolset): podklasy BaseToolset i ustawienia self._use_invocation_cache = False, dzięki którym ADK ponownie ocenia dostępne narzędzia natychmiast po zaktualizowaniu stanu sesji przez discover_zoo_mcp_servers, zamiast buforować początkowo pustą listę narzędzi. Gdy wywoływana jest funkcja get_tools(), odczytuje ona nazwy wykrytych serwerów ze stanu, tworzy uwierzytelniony obiekt McpToolset dla każdego serwera w wątku instancji roboczej i zwraca wszystkie aktywne narzędzia. Brak połączenia z jednym serwerem nie ma wpływu na pozostałe. _build_toolset() wykonuje blokujące operacje wejścia-wyjścia, więc _tools_of() uruchamia je w wątku instancji roboczej.
      class RegistryDiscoveredToolset(BaseToolset):
    def __init__(self, agent_registry: AgentRegistry):
        super().__init__()
        self._registry = agent_registry
        self._toolsets: dict[str, McpToolset] = {}
        self._use_invocation_cache = False

    def _build_toolset(self, server_name: str) -> McpToolset:
        header_provider = _GoogleAuthHeaders()
        server_registry = AgentRegistry(
            project_id=self._registry.project_id,
            location=self._registry.location,
            header_provider=header_provider,
        )
        toolset = server_registry.get_mcp_toolset(server_name)
        header_provider.bind(toolset.connection_params.url)
        toolset.connection_params.timeout = 30.0
        return toolset

    async def _tools_of(self, server_name: str, ctx: ReadonlyContext) -> list[BaseTool]:
        try:
            if server_name not in self._toolsets:
                self._toolsets[server_name] = await asyncio.to_thread(
                    self._build_toolset, server_name
                )
            return await self._toolsets[server_name].get_tools_with_prefix(ctx)
        except Exception as e:
            print(f"Skipping MCP server {server_name}: {e}")
            return []

    async def get_tools(self, readonly_context: ReadonlyContext | None = None) -> list[BaseTool]:
        if not readonly_context:
            return []
        record = _fresh_discovery(readonly_context.state)
        if not record or not record.get("servers"):
            if self._toolsets:
                await self.close()
            return []
        per_server = await asyncio.gather(
            *(self._tools_of(name, readonly_context) for name in record["servers"])
        )
        return [tool for tools in per_server for tool in tools]
    2. Dynamiczna instrukcja agenta (zoo_guide_instruction): ocenia stan sesji przed każdą turą modelu. Jeśli nie wykryto jeszcze żadnych serwerów, dołącza _DISCOVERY_PROMPT, aby poinstruować agenta, żeby najpierw wywołał discover_zoo_mcp_servers. Po pomyślnym wykryciu aktualizuje prompta, aby wyświetlić listę aktywnych serwerów MCP, i instruuje agenta, aby używał zamontowanych narzędzi zoo bezpośrednio, bez ponownego uruchamiania wykrywania:
      def zoo_guide_instruction(ctx: ReadonlyContext) -> str:
    record = _fresh_discovery(ctx.state)
    if record is None:
        return _ROLE + _DISCOVERY_PROMPT
    return _ROLE + (
        f"\nYour active Zoo MCP servers: {', '.join(record['servers'])}. "
        "Do not run discovery again; use the active zoo tools directly to fulfill the visitor's request."
    )

    Autoryzacja serwerów wykrywanych dynamicznie

    Wyszukiwanie semantyczne może zwrócić kilka serwerów, a ADK wywołuje dostawcę nagłówka, nie informując go, do którego serwera jest wysyłane żądanie. Dlatego _build_toolset() tworzy dedykowane _GoogleAuthHeaders i AgentRegistry dla każdego wykrytego serwera i wiąże dostawcę z punktem końcowym tego serwera. Dostawca jest taki sam jak w poprzednim kroku, ale ma flagę _bound, która powoduje, że nie działa, jeśli zostanie użyty przed bind().

    Logika przechowywania i odświeżania

    Aby uniknąć wysyłania zapytań do rejestru agentów przy każdej turze, a jednocześnie okresowo odświeżać wykryte punkty końcowe, agent.py implementuje przechowywanie stanu z czasem życia (TTL):
    DISCOVERY_STATE_KEY = "app:zoo_mcp_servers"
DISCOVERY_TTL_SECONDS = 24 * 60 * 60

def _fresh_discovery(state: Any) -> dict[str, Any] | None:
    if not state:
        return None
    record = state.get(DISCOVERY_STATE_KEY)
    if not record or not record.get("servers"):
        return None
    if time.time() - record.get("discovered_at", 0) > DISCOVERY_TTL_SECONDS:
        return None
    return record
    • Przechowywanie w zakresie aplikacji: wykryte nazwy serwerów i sygnatura czasowa discovered_at są przechowywane w stanie sesji pod kluczem app:zoo_mcp_servers. Użycie prefiksu app: ADK powoduje, że zakres zapisanego w pamięci podręcznej rekordu wykrywania obejmuje całe uruchomienie aplikacji.
    • Weryfikacja TTL i automatyczne odświeżanie: _fresh_discovery() sprawdza, czy zapisany rekord istnieje i czy jego wiek nie przekracza DISCOVERY_TTL_SECONDS (24 godzin). Gdy rekord jest aktualny, RegistryDiscoveredToolset ponownie używa połączeń McpToolset z pamięci podręcznej (self._toolsets), zoo_guide_instruction pomija prośbę o wykrycie, a każde powtarzające się wywołanie discover_zoo_mcp_servers natychmiast zwraca "already_discovered". Po upływie 24 godzin _fresh_discovery() zwraca None, co powoduje, że RegistryDiscoveredToolset.get_tools() zamyka istniejące sesje klienta MCP (await self.close()), a zoo_guide_instruction prosi agenta o przeprowadzenie nowego wykrywania.
  2. Uruchom agenta, aby przetestować proces wykrywania semantycznego. Użyj polecenia adk run:
    adk run zoo_agent
    lub uruchom interfejs internetowy ADK za pomocą polecenia adk web:
    adk web --port 8080
    a następnie otwórz w przeglądarce adres URL, który zostanie wyświetlony przez polecenie.Jeśli używasz Cloud Shell, dodaj argument --allow_origins, aby umożliwić otwieranie interfejsu internetowego z Cloud Shell:
    adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
    Gdy zadasz agentowi pytanie o zoo (np. How many animals are in the zoo?), zobaczysz, że najpierw wywołuje on discover_zoo_mcp_servers, dynamicznie montuje narzędzia z wykrytego serwera MCP, a następnie wywołuje narzędzia zoo, takie jak zoo_mcp_server_list_species i zoo_mcp_server_get_animals_by_species, aby odpowiedzieć na Twoje pytanie.Zadaj pytanie uzupełniające, np. Which is the oldest lion?. Agent odpowiada, korzystając z już zamontowanych narzędzi zoo, i nie wywołuje ponownie funkcji discover_zoo_mcp_servers, ponieważ rekord wykrywania w stanie sesji jest nadal aktualny.

7. Czyszczenie danych

Opcja 1. Usuń projekt

Jeśli na potrzeby tego ćwiczenia z programowania został przez Ciebie utworzony osobny projekt, najprostszym sposobem na uniknięcie płatności jest usunięcie projektu:

gcloud projects delete "${GOOGLE_CLOUD_PROJECT}" --quiet

Opcja 2. Usuwanie poszczególnych zasobów

Jeśli użyto istniejącego projektu i chcesz go zachować, usuń poszczególne zasoby utworzone w tym ćwiczeniu:

  1. Usuń rejestrację serwera MCP z rejestru agentów. Serwer został zarejestrowany ręcznie, więc usunięcie usługi Cloud Run nie spowoduje usunięcia rejestracji:
    gcloud agent-registry services delete zoo-mcp-server \
    --location=global \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  2. Usuń usługę serwera MCP Cloud Run:
    gcloud run services delete zoo-mcp-server \
    --region=us-west1 \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  3. Usuń konto usługi utworzone na potrzeby serwera MCP:
    gcloud iam service-accounts delete \
    "zoo-mcp-server-sa@${GOOGLE_CLOUD_PROJECT}.iam.gserviceaccount.com" \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  4. Jeśli repozytorium Artifact Registry cloud-run-source-deploy zostało utworzone specjalnie na potrzeby tego ćwiczenia z programowania, usuń je, aby uniknąć opłat za przechowywanie obrazów kontenerów:
    gcloud artifacts repositories delete cloud-run-source-deploy \
    --location=us-west1 \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  5. Opcjonalnie usuń lokalny katalog projektu utworzony w Cloud Shell:
    cd ~ && rm -rf adk-dynamic-discovery

8. Gratulacje

Z tego ćwiczenia wiesz już, jak dynamicznie wykrywać serwery MCP i korzystać z nich za pomocą Google Cloud Agent Registry i pakietu Agent Development Kit (ADK):

  1. Zarejestrowano serwer MCP: wdrożono zabezpieczony serwer MCP w Cloud Run i zarejestrowano go w rejestrze agentów.
  2. Wykrywanie serwera MCP według tożsamości: punkt końcowy serwera MCP jest rozpoznawany z rejestru agentów, a nie kodowany na stałe w adresie URL.
  3. Wykrywanie serwerów MCP semantycznie: umożliwia agentowi znajdowanie i montowanie narzędzi MCP w czasie działania za pomocą wyszukiwania w języku naturalnym, z zachowaniem i odświeżaniem.

Dalsze kroki