Dynamic MCP Server Discovery on Google Cloud

Why you need an MCP Server Registry #

Every agent that uses an MCP server needs the server’s URL. That seems harmless until the server moves to another region, gets redeployed under a new name or gets a separate staging copy. Then you have to update and redeploy every agent that has the URL hardcoded, and often nobody finds out until something breaks. With a few agents and a dozen servers, that one pasted URL becomes a real maintenance burden.

Here’s a typical way to connect an agent to an MCP server. It uses the Agent Development Kit (ADK) and its general-purpose McpToolset class, which works with any MCP server that uses the Streamable HTTP transport:

from google.adk.tools.mcp_tool.mcp_session_manager import (
    StreamableHTTPConnectionParams
)
from google.adk.tools.mcp_tool.mcp_toolset import (
    McpToolset
)

mcp_toolset = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url=MCP_SERVER_URL,
        timeout=30.0,  # custom timeout
    ),
)

McpToolset handles the MCP protocol for the agent: the connection, listing the server’s tools and calling them. If the server requires authentication, a header provider can add headers such as Authorization to each request. ADK also has higher-level, ready-made integrations for some MCP servers, including several Google Cloud ones. I use the general-purpose toolset here because it works the same way with any server, including custom ones that you or your team build. Once the toolset is created, you can add it to an ADK agent as a tool.

from google.adk.agents.llm_agent import Agent

_ROLE = """You are the friendly guide.
Your goal is to help users with their requests.
Rules:
- Never answer questions from your own knowledge.
  Every fact MUST come from the tools. If the tools return nothing, say so.
- Use the tools on the very first question as well. Do not reply with a
  generic greeting when the user has already asked something.
- If the user only greets you, briefly introduce yourself and offer to
  help them
"""

agent = Agent(
    model="gemini-3.8-flash",
    name="my_agent",
    description="The friendly assistant for handling a user's requests.",
    instruction=_ROLE,
    tools=[
        mcp_toolset,
    ],
)

Other agent frameworks have their own ways to connect to MCP servers, and they all assume the same thing: you know the server’s URL before the agent starts. In practice, you either put the URL into the deployment or fetch it from “somewhere” before you configure the agent.

That “somewhere” is what this post is about. Instead of every agent keeping its own copy of the URL, you store it once in a registry, and agents look it up when they need it. The registry becomes the one place that records which servers exist and where they run. Agents only need to know what they’re looking for. Many services already offer this kind of registry for MCP servers.

The rest of this post shows how it works with Google Cloud’s version. In June 2026, Google Cloud made Agent Registry generally available. Despite its name, Agent Registry is also an MCP server registry. It handles registration and discovery for several types of AI workloads: MCP servers, agents, inference services (LLM engines) and skills. I’ll focus on discovering MCP servers, but I plan to explore Agent Registry more in the future.

Direct approach #

When I built my first implementation of the agent that uses Agent Registry, I took a straightforward approach: walk through all registered MCP servers and search for mine. ADK provides the AgentRegistry class, a client for the Agent Registry API. Conveniently, the class can build an McpToolset instance from the registration record. My code looked like this:

from google.adk.integrations.agent_registry import (
    AgentRegistry
)

registry = AgentRegistry(
    project_id=PROJECT_ID,
    location=LOCATION,
)

mcp_toolset = None
mcp_servers = registry.list_mcp_servers().get("mcpServers", [])
for s in mcp_servers:
    if s.get("mcpServerId") == MCP_SERVER_URN:
        mcp_toolset = registry.get_mcp_toolset(mcp_server_name=s["name"])
        break

You can add mcp_toolset to the agent’s tools as shown in the previous section. The code no longer needs the server’s URL. However, to find the server, the code uses its URN ‒ a uniform resource name that Agent Registry uses as the server’s unique ID. For MCP servers, the URN has the following format:

urn:mcp:projects-PROJECT1_NUMBER:projects:PROJECT2_NUMBER:locations:LOC:SERVICE

The variable parts of the URN are:

• PROJECT1_NUMBER : the project number of the Google Cloud project where the MCP server is registered.
• PROJECT2_NUMBER : the project number of the Google Cloud runtime where the MCP server is deployed. When the MCP server is registered through the Agent Registry API,PROJECT2_NUMBER is the same asPROJECT1_NUMBER.
• LOC : the Agent Registrylocation that was used to register the MCP server.
• SERVICE : the path to the service.

Agent Registry integrates with Cloud Run and GKE to register an MCP server automatically when it’s deployed to one of these runtimes. Automatic registration has a few caveats, which I cover in the “Navigating the pitfalls” section. The structure of SERVICE depends on how the server was registered:

• NAME stands for the service or deployment name.
• CLUSTER_NAME stands for the name of the GKE cluster.
• NAMESPACE stands for the Kubernetes namespace of the MCP server deployment.

At first glance, this solution just replaces one piece of information (the URL) with another (the URN). The difference is that the agent no longer stores the endpoint; it reads the endpoint from the registry. This decouples the MCP server’s deployment details from the development lifecycle of the agents that use it.

Fully dynamic discovery #

Knowing the URN is still a burden, especially in situations where the agent might not be aware of any registration details. Additionally, these details may change during the lifetime of the agent, making discovery at launch inflexible.

Agent Registry also supports semantic search for MCP servers. By combining it with a custom tool, a custom toolset and dynamic instructions, you can find the best-suited MCP server at runtime. My implementation does everything in a single agent. Depending on the number of MCP servers and the size of the context, you may want to split this work across several agents.

Semantic search

The AgentRegistry ADK class has a search_mcp_servers method that supports both keyword and semantic search. The following code wraps the call to the method in a tool, so the agent can compose a semantic query and run it against Agent Registry.

import time
from typing import Any
from google.adk.tools import ToolContext

def discover_mcp_servers(
    query: str,
    ctx: ToolContext,
) -> dict[str, Any]:
    cached = _fresh_discovery(ctx.state)
    if cached:
        return {
            "status": "already_discovered",
            "message": "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 servers found matching '{query}'.",
        }
    servers = [s["name"] for s in found_servers if "name" in s]
    ctx.state[DISCOVERY_STATE_KEY] = {
        "servers": servers,
        "discovered_at": time.time(),
    }
    return {
        "status": "discovered",
        "message": f"Successfully discovered {len(servers)} MCP server(s).",
        "servers": servers,
    }

The code uses a registry variable that is initialized the same way as in the direct approach. But instead of going through the list and matching against known values, the code searches for the MCP server with a query that the agent composes. The code stores the results in the state under the DISCOVERY_STATE_KEY key. See ADK state scopes for ways to control the scope of the discovery results. The code also sets a retention timeout, in case the registration changes while the agent is still running. The _fresh_discovery function returns the record from the state if it exists and hasn’t expired:

import time
from typing import Any

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

Discovering and using the tools

The code above discovers servers, not their tools. Although the tool manifest is included in the Agent Registry response, it isn’t enough to call the tools. The obvious way would be to create an instance of McpToolset and add it to the agent’s tools. However, that would require re-creating the agent at runtime. It’s possible, but in my opinion the biggest drawback is the added complexity of implementing and maintaining the agent’s code, especially if the application doesn’t otherwise need dynamically created agents. A simpler way is a custom toolset class that creates toolsets for the discovered servers and returns their tools at runtime. ADK provides the BaseToolset class. A custom toolset inherits from it and implements the get_tools method, which returns the tools of the discovered MCP servers:

import asyncio
from google.adk.agents.readonly_context import (
    ReadonlyContext
)
from google.adk.integrations.agent_registry import (
    AgentRegistry
)
from google.adk.tools import BaseTool
from google.adk.tools.base_toolset import BaseToolset
from google.adk.tools.mcp_tool.mcp_toolset import (
    McpToolset
)

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

    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._registry.get_mcp_toolset, server_name
                )
                self._toolsets[server_name].connection_params.timeout = 30.0
            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,
        ctx: ReadonlyContext | None = None,
    ) -> list[BaseTool]:
        if not ctx:
            return []
        record = _fresh_discovery(ctx.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, ctx) for name in record["servers"])
        )
        return [tool for tools in per_server for tool in tools]

    async def close(self) -> None:
        if self._toolsets:
            await asyncio.gather(
                *(toolset.close() for toolset in self._toolsets.values()),
                return_exceptions=True,
            )
            self._toolsets.clear()

Dynamic instructions

The agent’s instructions have to reflect whether the MCP servers are already discovered. In ADK, an agent’s instructions can be static text or the output of a function. The following function tells the agent to discover tools first and to use them once they’re discovered:

from google.adk.agents.readonly_context import (
    ReadonlyContext
)

_ROLE = """You are a friendly, knowledgeable guide. You help users to find answers to their inquires. Base every fact about on tool results; never invent them.
"""
def agent_instructions(ctx: ReadonlyContext) -> str:
    record = _fresh_discovery(ctx.state)
    if record is None:
        return _ROLE + """TOOL DISCOVERY - do this once before answering the user:
1. Call `discover_mcp_servers` with a semantic query.
2. As soon as the discovery succeeds, the tools become available to you.
3. Immediately use those tools to answer the user's request.
"""
    return _ROLE + (
        f"\nYour active MCP servers: {', '.join(record['servers'])}. "
        "Do not run discovery again; use the active tools directly to fulfill the user's request."
    )

What’s left is to update the code that creates the agent so that it uses all these constructs:

from google.adk.agents.llm_agent import Agent

agent = Agent(
    model="gemini-3.8-flash",
    name="my_agent",
    description="The friendly assistant for handling a user's requests.",
    instruction=agent_instructions,
    tools=[
        discover_mcp_servers,
        RegistryDiscoveredToolset(registry),
    ],
)

Navigating the pitfalls #

The path to this result wasn’t without friction. Here are a few pitfalls I encountered and how you can work around them:

Location dependency: You may have noticed that theAgentRegistry constructor accepts a location argument. You see only the servers registered in the location that you pass in this argument. For example, if you registered your server inus-central1 but passedglobal as the location argument, you won’t find the server. The good news is that the registration location doesn’t have to match the deployment location.Tip Unless data residency requirements apply to you, register your MCP servers in the global location. #

Automatic registration with Cloud Run and GKE: As mentioned, GKE and Cloud Run can automatically register workloads labeled as MCP servers. These registrations are performed in the same project and region as the workload. For example, deploying an MCP server on GKE in theus-west1 region registers it in theus-west1 Agent Registry location.Warning When I wrote this post, automatic registration in Cloud Run recorded only the server’s name and type, which made the record unusable for discovery. Verify the registered metadata, such as the tool manifest, before you use it in production. Tip Register your MCP servers explicitly to control the registration location. #

Two project numbers in URN: The URN format requires two separate project numbers in the URN prefix:

urn:mcp:projects-PROJECT1_NUMBER:projects:PROJECT2_NUMBER:...

PROJECT1_NUMBER represents the publisher, in this case Agent Registry.PROJECT2_NUMBER represents the runtime resource. When an MCP server is manually registered using the Agent Registry API, the two values are the same. The two numbers can differ only when Agent Registry ingests or indexes native Google Cloud runtimes across multiple projects (such as Cloud Run, Vertex AI / Reasoning Engine, or GKE).Tip If you use the URN to find the MCP server, make sure you know both project numbers: the Agent Registry project and the server’s runtime project. Don’t assume they’re always the same. #

Tokens for authorized invocations: The code snippets include commented-out lines that show where to add authorization headers. It’s important to distinguish between the tokens that you use to call Agent Registry and the tokens that you use to call MCP servers. Reusing one token for both types of calls, or for calls to different MCP servers, can create a security vulnerability or simply not work.Tip Use the agent’s Application Default Credentials to call Agent Registry, and use a dedicated mechanism to obtain tokens for calls to MCP servers. If you use RegistryDiscoveredToolset with servers that need different credentials, give each server its ownAgentRegistry instance and header provider. See theDynamic MCP Server Discovery codelab for an example.

Alternatives and final thoughts #

The implementation I went with is only one way to use Agent Registry for discovery. Another way is to connect your agent to Agent Registry’s own MCP server (in Preview at the time of writing) instead of writing a custom discovery tool. That server has a fixed, Google-managed endpoint, so hardcoding its URL doesn’t bring back the problem this post started with. If you build an agent whose job is tool discovery, the Agent Registry MCP server gives you discovery without writing and maintaining your own tool. Keep in mind that its search tool performs keyword or prefix search, not the semantic search that I used in this post.

Whichever option you choose, your agents no longer carry the server’s URL. In my implementation, the agent is equipped with two tools: discover_mcp_servers to discover MCP server(s) and RegistryDiscoveredToolset to mount the tools of the discovered servers, and use them to answer the user’s requests. Later requests reuse the discovery until it expires. When a server moves to another region or gets redeployed under a new name, you update its registration once. The agents find it at the new location after their cached discovery expires, with no code change and no redeployment.

You can find the complete code, including the authorization and error handling that I left out of the snippets, in the Dynamic MCP Server Discovery with Agent Registry and ADK codelab. The code for all steps of the codelab is on GitHub.