Découverte dynamique de serveurs MCP avec Agent Registry et ADK

1. Introduction

Dans cet atelier de programmation, vous allez apprendre à utiliser Google Cloud Agent Registry et l'Agent Development Kit (ADK) pour découvrir et appeler de manière dynamique des serveurs MCP (Model Context Protocol) enregistrés au moment de l'exécution, ce qui élimine les points de terminaison et les configurations d'outils codés en dur.

Cet atelier utilise ADK et le langage de programmation Python pour créer une application. Les mêmes principes peuvent être appliqués dans d'autres langages de programmation et à l'aide d'autres frameworks de développement d'agents.

Objectifs de l'atelier

  • Activer et configurer Google Cloud Agent Registry dans votre projet
  • Déployer un serveur MCP sur Cloud Run et l'enregistrer dans Agent Registry
  • Créer un agent ADK qui appelle le serveur MCP à l'aide d'une URL codée en dur
  • Modifier l'agent pour qu'il découvre le serveur MCP dans Agent Registry par son identité
  • Modifier l'agent pour qu'il découvre les serveurs MCP au moment de l'exécution à l'aide de la recherche sémantique

Prérequis

  • Un navigateur Web récent (tel que Chrome)
  • Un projet Google Cloud avec facturation activée
  • Un compte disposant des autorisations nécessaires pour créer et gérer des ressources, et configurer des stratégies IAM sur le projet Google Cloud

2. Configurer votre environnement

Les instructions de cet atelier de programmation supposent que vous exécutez des commandes dans Cloud Shell dans la console Google Cloud. Cloud Shell est un environnement de ligne de commande exécuté dans Google Cloud et fourni avec les outils nécessaires. Si vous avez installé la CLI gcloud, git et Python 3 dans votre environnement local, vous pouvez également y exécuter les commandes.

Créer un projet Google Cloud

  1. Dans la console Google Cloud, sur la page de sélection du projet, créez un projet Google Cloud.
  2. Assurez-vous que la facturation est activée pour votre projet Cloud. Découvrez comment vérifier si la facturation est activée sur un projet.

Démarrer un terminal Cloud Shell

  1. À l'aide du sélecteur de ressources, sélectionnez le projet Google Cloud que vous avez créé.
  2. Cliquez sur Activer Cloud Shell Icône Activer Cloud Shell en haut de la console Google Cloud.
  3. Une fois connecté à Cloud Shell, vérifiez votre authentification :
    gcloud auth list
  4. Vérifiez que votre projet est configuré :
    gcloud config get project
  5. Si votre projet n'est pas défini comme prévu ou si vous utilisez un environnement sur votre ordinateur local au lieu de Cloud Shell, configurez l'ID du projet :
    export GOOGLE_CLOUD_PROJECT=<YOUR_PROJECT_ID>
gcloud config set project "${GOOGLE_CLOUD_PROJECT}"

Sélectionner un projet Google Cloud et définir des variables d'environnement

  1. Activez les API Google Cloud requises :
    gcloud services enable \
  agentregistry.googleapis.com \
  aiplatform.googleapis.com \
  cloudbuild.googleapis.com \
  run.googleapis.com \
  --project="${GOOGLE_CLOUD_PROJECT}"

Configurer l'environnement Python

  1. Créez et activez un environnement virtuel, puis installez les dépendances requises :
    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. Créer et enregistrer un serveur MCP Zoo

Au cours de cette étape, vous allez créer et enregistrer un serveur MCP Zoo. Le serveur expose trois outils : list_species (lister toutes les espèces qui vivent dans le zoo), get_animals_by_species (lister les animaux d'une espèce spécifique) et get_animal_details (obtenir des informations supplémentaires sur un animal spécifique).

Vous trouverez tous les fichiers de cette étape sur GitHub.

Créer le serveur MCP

  1. Configurer un dossier et un projet Python pour le serveur 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. Téléchargez le code du serveur 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

Déployer le serveur MCP sur Cloud Run et l'enregistrer dans Agent Registry

Vous allez ensuite créer Dockerfile pour le serveur et déployer le serveur sur Cloud Run.

  1. Créez 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. Créez un compte de service désigné pour le serveur :
    gcloud iam service-accounts create zoo-mcp-server-sa \
    --display-name="Zoo MCP Server Service Account"
  3. Déployez le serveur dans la région 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
    Si l'invite suivante s'affiche, confirmez la création d'un dépôt Artifact Registry. Cette invite s'affiche si vous n'avez pas de dépôt Artifact Registry dans votre projet pour la région us-west1 :
    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. Enregistrez le serveur MCP dans Agent Registry à l'emplacement "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"
    Ces commandes génèrent toolspec.json avec le fichier manifeste de l'outil et appellent l'API Agent Registry pour enregistrer le serveur.
  5. Vérifiez que l'agent apparaît dans le registre :
    gcloud agent-registry mcp-servers list \
    --location=global \
    --filter="display_name=zoo-mcp-server"
    Vous verrez une sortie YAML contenant les informations du serveur MCP, comme suit :
    ---
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. Créer un agent ADK de base pour utiliser le serveur MCP Zoo

Dans cette étape, vous allez créer un agent ADK pour répondre aux requêtes des utilisateurs concernant le zoo à l'aide de l'URL codée en dur du serveur MCP du zoo. La méthode codée en dur vous oblige à connaître l'URL du serveur MCP AVANT le déploiement de l'agent.

Vous trouverez les fichiers sources de cette étape sur GitHub.

Créer l'application d'agent

Créez un échafaudage du projet ADK et de l'agent Zoo qui utilise le serveur MCP Zoo.

  1. Configurez un dossier et un projet pour l'agent 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. Créez le dossier du package de l'agent et téléchargez le fichier __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. Créez le fichier .env et renseignez-le avec les variables d'environnement utilisées dans cette étape et les suivantes :
    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
    Cette commande écrit toutes les variables d'environnement requises pour cette étape et les étapes de découverte suivantes dans zoo_agent/.env. Vous n'avez donc pas besoin de recréer le fichier ultérieurement. Dans cette étape, les variables suivantes sont utilisées :
    • GOOGLE_GENAI_USE_ENTERPRISE, GOOGLE_CLOUD_PROJECT et GOOGLE_CLOUD_LOCATION : utilisés implicitement par l'ADK et le SDK Google GenAI pour acheminer les appels au modèle Gemini via Vertex AI dans votre projet et votre emplacement Google Cloud (global). Ils stockent également l'emplacement du registre d'agents où le serveur MCP est enregistré.
    • GOOGLE_CLOUD_PROJECT_NUMBER : non utilisé dans cette étape. Il est prérempli ici pour être utilisé à l'étape suivante.
    • MCP_SERVER_URL : lisez directement dans zoo_agent/agent.py pour configurer McpToolset avec l'URL Cloud Run codée en dur du serveur MCP du zoo.
  4. Téléchargez agent.py avec le code de l'agent Zoo qui se connecte au serveur 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

Fonctionnement de l'autorisation des serveurs MCP

Comme le service Cloud Run zoo-mcp-server a été déployé avec les options --invoker-iam-check et --no-allow-unauthenticated, chaque requête HTTP adressée au serveur MCP doit inclure un jeton d'ID signé par Google dans l'en-tête Authorization. Le jeton doit identifier un appelant disposant de l'autorisation run.routes.invoke sur le service.

Dans zoo_agent/agent.py, la classe _GoogleAuthHeaders obtient les identifiants du jeton d'ID de deux manières. Il appelle d'abord fetch_id_token_credentials() pour obtenir un jeton d'ID pour l'URL du service à partir du serveur de métadonnées ou d'une clé de compte de service. En cas d'échec, il utilise le jeton d'ID des identifiants par défaut de l'application (ADC) de votre utilisateur. Chaque fois que l'instance est appelée, elle actualise le jeton si nécessaire et renvoie l'en-tête 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}"}

L'agent transmet une instance de la classe à McpToolset en tant qu'argument header_provider. L'ADK appelle le fournisseur d'en-têtes et associe les en-têtes qu'il renvoie à chaque requête adressée au serveur MCP. Comme le client HTTP est créé par ADK, le même code fonctionne quelle que soit la version du SDK Python MCP et de sa bibliothèque HTTP installée :

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

L'audience demandée est l'URL du service (MCP_SERVER_URL), tandis que l'URL de connexion inclut le chemin d'accès /mcp où le serveur FastMCP diffuse le protocole MCP. Le délai avant expiration de 30 secondes laisse de la place pour les démarrages à froid de Cloud Run.

Tester l'agent à l'aide du playground ADK dans votre environnement local

Avant de tester l'agent, assurez-vous que votre compte utilisateur peut appeler le serveur MCP.

  1. Terminal local uniquement (ignorez cette étape dans Cloud Shell) : initialisez les ADC avec vos identifiants utilisateur :
    gcloud auth application-default login
    Suivez les instructions pour terminer l'initialisation des ADC.
  2. (Facultatif) Accordez à votre compte utilisateur l'autorisation d'appeler le serveur MCP distant :
    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. Exécutez votre agent avec la commande adk run :
    adk run zoo_agent
    ou exécutez l'interface Web ADK avec la commande adk web :
    adk web --port 8080
    , puis ouvrez l'URL affichée par la commande dans le navigateur.Si vous utilisez Cloud Shell, ajoutez l'argument --allow_origins pour permettre l'ouverture de l'interface Web depuis Cloud Shell :
    adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
    Après avoir exécuté la commande, recherchez l'icône Aperçu sur le WebIcône Aperçu sur le Web dans la barre d'outils Cloud Shell (à droite). Cliquez dessus et sélectionnez Prévisualiser sur le port 8080. Cloud Shell ouvre un nouvel onglet de navigateur affichant l'interface Web ADK.
  4. Commencez à discuter avec l'agent. Par exemple, posez-lui des questions sur les animaux du zoo :
    How many animals are in the zoo?
    L'agent appelle list_species, puis get_animals_by_species pour chaque espèce afin de calculer la réponse.

L'interface Web ADK vous permet d'explorer le fonctionnement de l'agent, y compris ses appels au serveur MCP du zoo. Pour savoir comment procéder, consultez la documentation sur l'UI Web.

5. Découvrir dynamiquement le point de terminaison du serveur MCP

Dans cette étape, vous allez modifier l'agent pour qu'il découvre le point de terminaison du serveur MCP Zoo enregistré dans Agent Registry au lieu de coder en dur son URL de point de terminaison.

Vous trouverez tous les fichiers de cette étape sur GitHub.

Vous utilisez l'intégration d'ADK à Agent Registry pour rechercher et récupérer l'URL du serveur. Pour l'implémenter, vous avez besoin des informations suivantes :

  • Numéro de projet et emplacement du registre d'agents où votre serveur MCP est enregistré (global).
  • ID de service que vous avez utilisé pour enregistrer le serveur MCP (zoo-mcp-server).

Mettre à jour l'agent pour qu'il utilise Agent Registry

  1. Téléchargez le fichier agent.py mis à jour depuis GitHub pour remplacer la première version de l'agent :
    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

    Variables d'environnement dans cette étape

    Avec cette mise à jour, l'agent s'appuie sur les variables d'environnement suivantes de zoo_agent/.env :
    • GOOGLE_CLOUD_PROJECT_NUMBER : désormais requis pour construire l'URN d'identité du serveur MCP (MCP_SERVER_URN).
    • GOOGLE_CLOUD_PROJECT, GOOGLE_GENAI_USE_ENTERPRISE et GOOGLE_CLOUD_LOCATION : utilisés par l'ADK pour configurer l'accès à Vertex AI. L'emplacement est également utilisé pour détecter le serveur MCP.
    • Plus nécessaire : MCP_SERVER_URL n'est plus utilisé, car l'agent récupère l'URL du point de terminaison du serveur MCP à partir d'Agent Registry plutôt que de la lire depuis l'environnement.

    Fonctionnement de l'identité et de la découverte des serveurs MCP dans AgentRegistry

    Dans le registre d'agents, chaque serveur MCP enregistré possède un identifiant unique stocké dans son champ mcpServerId, au format URN (Uniform Resource Name). Vous pouvez le voir dans le résultat de la commande gcloud agent-registry mcp-servers list à l'étape 3. L'agent reconstruit cet identifiant à partir des variables d'environnement :
    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"
    L'URN encode le numéro de projet du registre (projects-{PROJECT_NUMBER}), le numéro de projet et l'emplacement où le serveur est enregistré (projects:{PROJECT_NUMBER}:locations:{LOCATION}), ainsi que l'ID de service que vous avez utilisé dans la commande gcloud agent-registry services create (agentregistry:services:zoo-mcp-server).L'agent liste les serveurs MCP enregistrés avec registry.list_mcp_servers() et compare l'mcpServerId de chaque serveur à MCP_SERVER_URN. Il transmet ensuite le name de la ressource du serveur correspondant (projects/.../locations/.../mcpServers/...) à registry.get_mcp_toolset(), qui lit l'URL du point de terminaison à partir de l'interface enregistrée du serveur (https://...run.app/mcp) et renvoie un McpToolset connecté à cette 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() ajoute le nom à afficher du serveur au début des noms des outils. L'agent voit donc les outils sous les noms zoo_mcp_server_list_species, zoo_mcp_server_get_animals_by_species et zoo_mcp_server_get_animal_details. Le préfixe évite les conflits de noms lorsqu'un agent utilise des outils de plusieurs serveurs MCP. La dernière ligne augmente le délai avant expiration de la connexion par rapport à sa valeur par défaut de 5 secondes pour laisser de la place aux démarrages à froid de Cloud Run.

    Comment l'autorisation change avec AgentRegistry

    AgentRegistry authentifie ses propres appels d'API, comme list_mcp_servers(), avec ADC. Le fournisseur d'en-tête n'est donc nécessaire que pour les appels au serveur MCP découvert. Comme AgentRegistry crée le McpToolset en interne, vous transmettez le fournisseur d'en-tête au constructeur AgentRegistry.À l'étape précédente, l'audience était la valeur MCP_SERVER_URL que vous connaissiez à l'avance. Désormais, l'URL n'est connue qu'après la découverte. ADK appelle le fournisseur d'en-tête lorsque l'agent envoie une requête au serveur MCP, et non lorsque AgentRegistry est créé. Cela permet à l'agent de créer d'abord le fournisseur, puis de l'associer au point de terminaison découvert après le renvoi de get_mcp_toolset(). La méthode bind() obtient les identifiants de la même manière que le constructeur à l'étape précédente, l'audience étant extraite de l'URL du point de terminaison découvert sans le chemin d'accès /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
    La méthode __call__ est identique à celle de l'étape précédente, sauf qu'elle génère une erreur si elle est appelée avant bind().
  2. Exécutez votre agent pour tester la modification. Exécutez la commande adk run :
    adk run zoo_agent
    ou exécutez l'interface Web ADK avec la commande adk web :
    adk web --port 8080
    , puis ouvrez l'URL affichée par la commande dans le navigateur.Si vous utilisez Cloud Shell, ajoutez l'argument --allow_origins pour permettre l'ouverture de l'interface Web depuis Cloud Shell :
    adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
    Posez à l'agent la même question qu'à l'étape précédente. La réponse est la même, mais dans l'interface Web ADK, les appels d'outils s'affichent désormais sous la forme zoo_mcp_server_list_species et zoo_mcp_server_get_animals_by_species.

Cette modification vous a permis de ne plus avoir besoin de l'URL explicite du serveur MCP. Toutefois, vous devez toujours connaître le numéro de projet, l'emplacement du registre et l'ID de service utilisé pour enregistrer le serveur MCP. De plus, la découverte est effectuée de manière statique lors de l'initialisation de l'agent.

6. Utiliser la recherche sémantique pour le serveur MCP dans la découverte dynamique

Dans cette étape, vous allez améliorer la découverte dynamique du serveur MCP en remplaçant la correspondance exacte des URN au démarrage par une recherche sémantique au moment de l'exécution. Cette approche élimine la nécessité de connaître le numéro du projet dans lequel le serveur est déployé ou l'identité exacte du serveur (MCP_SERVER_URN). Elle intègre également la découverte des outils directement dans le workflow d'exécution standard de l'agent, au lieu de la gérer lors de l'initialisation.

Vous trouverez tous les fichiers de cette étape sur GitHub.

Mettre à jour l'agent pour qu'il utilise la découverte sémantique au moment de l'exécution

  1. Téléchargez le fichier agent.py mis à jour depuis GitHub pour écraser l'implémentation existante :
    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

    Variables d'environnement dans cette étape

    Comme l'agent ne crée plus d'URN de serveur MCP, la configuration requise est encore plus simple :
    • GOOGLE_CLOUD_PROJECT, GOOGLE_GENAI_USE_ENTERPRISE et GOOGLE_CLOUD_LOCATION : utilisés par l'ADK pour configurer l'accès à Vertex AI. L'emplacement est également utilisé pour détecter le serveur MCP.
    • Plus requis : GOOGLE_CLOUD_PROJECT_NUMBER n'est plus nécessaire (en plus de MCP_SERVER_URL, qui a déjà été supprimé à l'étape précédente).
    Au lieu d'appeler registry.list_mcp_servers() lors de l'initialisation du module et de comparer chaque entrée à un MCP_SERVER_URN construit, agent.py définit désormais une fonction d'outil de découverte discover_zoo_mcp_servers qui appelle registry.search_mcp_servers() avec 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,
    }
    Cela permet à l'agent de rechercher dans Agent Registry au moment de l'exécution à l'aide d'une description en langage naturel des fonctionnalités dont il a besoin (par exemple, "zoo animals, exhibits, habitat, care, feeding schedule") et d'enregistrer les noms des ressources de serveur découvertes dans l'état de session ADK (tool_context.state).

    Comment la découverte dynamique est intégrée au workflow de l'agent

    Pour installer des outils à partir de serveurs découverts au moment de l'exécution et guider l'agent tout au long du flux de découverte, puis d'appel, agent.py combine une implémentation BaseToolset personnalisée (RegistryDiscoveredToolset) avec un fournisseur d'instructions dynamique (zoo_guide_instruction) :
    1. Ensemble d'outils dynamique (RegistryDiscoveredToolset) : sous-classe BaseToolset et définit self._use_invocation_cache = False pour qu'ADK réévalue les outils disponibles immédiatement après que discover_zoo_mcp_servers a mis à jour l'état de la session, au lieu de mettre en cache une liste d'outils initialement vide. Lorsque get_tools() est appelé, il lit les noms des serveurs découverts à partir de l'état, crée un McpToolset authentifié pour chaque serveur dans un thread de travail et renvoie tous les outils actifs. L'impossibilité d'accéder à un serveur n'affecte pas les autres. _build_toolset() effectue des E/S bloquantes, donc _tools_of() l'exécute dans un thread de travail.
      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. Instruction d'agent dynamique (zoo_guide_instruction) : évalue l'état de la session avant chaque tour de modèle. Lorsqu'aucun serveur n'a encore été détecté, il ajoute _DISCOVERY_PROMPT pour indiquer à l'agent d'appeler d'abord discover_zoo_mcp_servers. Une fois la découverte réussie, il met à jour la requête pour lister les serveurs MCP actifs et demande à l'agent d'utiliser directement les outils du zoo installés sans relancer la découverte :
      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."
    )

    Autorisation pour les serveurs détectés de manière dynamique

    La recherche sémantique peut renvoyer plusieurs serveurs, et l'ADK appelle un fournisseur d'en-têtes sans lui indiquer vers quel serveur une requête est envoyée. Par conséquent, _build_toolset() crée un _GoogleAuthHeaders et un AgentRegistry dédiés pour chaque serveur découvert, et lie le fournisseur au point de terminaison de ce serveur. Le fournisseur est le même qu'à l'étape précédente, avec un flag _bound qui le fait échouer s'il est utilisé avant bind().

    Logique de conservation et d'actualisation

    Pour éviter d'interroger le registre d'agents à chaque tour tout en actualisant régulièrement les points de terminaison découverts, agent.py implémente la conservation de l'état avec une valeur TTL (Time To Live) :
    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
    • Conservation au niveau de l'application : les noms de serveurs découverts et un code temporel discovered_at sont stockés dans l'état de la session sous la clé app:zoo_mcp_servers. L'utilisation du préfixe app: d'ADK permet de limiter la portée de l'enregistrement de découverte mis en cache à l'exécution de l'application.
    • Validation de la valeur TTL et actualisation automatique : _fresh_discovery() vérifie que l'enregistrement enregistré existe et que son âge ne dépasse pas DISCOVERY_TTL_SECONDS (24 heures). Tant que l'enregistrement est récent, RegistryDiscoveredToolset réutilise les connexions McpToolset mises en cache (self._toolsets), zoo_guide_instruction omet l'invite de découverte et tout appel répété à discover_zoo_mcp_servers renvoie immédiatement "already_discovered". Une fois le TTL de 24 heures expiré, _fresh_discovery() renvoie None, ce qui amène RegistryDiscoveredToolset.get_tools() à fermer les sessions client MCP existantes (await self.close()) et zoo_guide_instruction à inviter l'agent à effectuer une nouvelle découverte.
  2. Exécutez votre agent pour tester le workflow de découverte sémantique. Exécutez la commande adk run :
    adk run zoo_agent
    ou exécutez l'interface Web ADK avec la commande adk web :
    adk web --port 8080
    , puis ouvrez l'URL affichée par la commande dans le navigateur.Si vous utilisez Cloud Shell, ajoutez l'argument --allow_origins pour permettre l'ouverture de l'interface Web depuis Cloud Shell :
    adk web --port 8080 --allow_origins="regex:https://.*\.cloudshell\.dev"
    Lorsque vous posez une question sur le zoo à l'agent (par exemple, How many animals are in the zoo?), vous pouvez observer qu'il appelle d'abord discover_zoo_mcp_servers, installe de manière dynamique les outils du serveur MCP découvert, puis appelle les outils du zoo, tels que zoo_mcp_server_list_species et zoo_mcp_server_get_animals_by_species, pour répondre à votre question.Posez une question complémentaire, par exemple Which is the oldest lion?. L'agent répond à l'aide des outils de zoo déjà installés et n'appelle pas à nouveau discover_zoo_mcp_servers, car l'enregistrement de découverte dans l'état de la session est toujours récent.

7. Effectuer un nettoyage

Option 1 : supprimer le projet

Si vous avez créé un projet dédié pour cet atelier de programmation, le moyen le plus simple d'éviter la facturation consiste à supprimer le projet :

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

Option 2 : Supprimer des ressources individuelles

Si vous avez utilisé un projet existant et que vous souhaitez le conserver, supprimez les ressources individuelles créées lors de cet atelier de programmation :

  1. Supprimez l'enregistrement du serveur MCP d'Agent Registry. Comme vous avez enregistré le serveur manuellement, la suppression du service Cloud Run ne supprime pas l'enregistrement :
    gcloud agent-registry services delete zoo-mcp-server \
    --location=global \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  2. Supprimez le service de serveur MCP Cloud Run :
    gcloud run services delete zoo-mcp-server \
    --region=us-west1 \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  3. Supprimez le compte de service créé pour le serveur MCP :
    gcloud iam service-accounts delete \
    "zoo-mcp-server-sa@${GOOGLE_CLOUD_PROJECT}.iam.gserviceaccount.com" \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  4. Si le dépôt Artifact Registry cloud-run-source-deploy a été créé spécifiquement pour cet atelier de programmation, supprimez-le pour éviter des frais de stockage d'images de conteneurs :
    gcloud artifacts repositories delete cloud-run-source-deploy \
    --location=us-west1 \
    --project="${GOOGLE_CLOUD_PROJECT}" \
    --quiet
  5. Vous pouvez également supprimer le répertoire du projet local créé dans Cloud Shell :
    cd ~ && rm -rf adk-dynamic-discovery

8. Félicitations

Dans cet atelier de programmation, vous avez appris à découvrir et à utiliser de manière dynamique des serveurs MCP avec Google Cloud Agent Registry et l'Agent Development Kit (ADK) :

  1. Enregistrer un serveur MCP : vous avez déployé un serveur MCP sécurisé sur Cloud Run et l'avez enregistré dans Agent Registry.
  2. Découvert un serveur MCP par identité : vous avez résolu le point de terminaison du serveur MCP à partir d'Agent Registry au lieu de coder son URL en dur.
  3. Découverte sémantique des serveurs MCP : permettez à l'agent de trouver et d'installer des outils MCP au moment de l'exécution à l'aide de la recherche en langage naturel, avec conservation et actualisation.

Étapes suivantes