A guide to Google ADK and MCP integration with an external server

For AI-powered agents to perform useful, real-world tasks, they need to reliably access tools and up-to-the-minute information that lives outside the base model. Anthropic’s Model Context Protocol (MCP) is designed to address this, providing a standardized way for agents to retrieve that crucial, external context needed to inform their responses and actions.

This is vital for developers who need to build and deploy sophisticated agents that can leverage enterprise data or public tools. But integrating agents built with Google's Agent Development Kit (ADK) to communicate effectively with an MCP server, especially one hosted externally, might present some integration challenges.

Today, we’ll guide you through developing ADK agents that connect to external MCP servers, initially using Server-Sent Events (SSE). We’ll take an example of an ADK agent leveraging MCP to access Wikipedia articles, which is a common use case to retrieve external specialised data. We will also introduce Streamable HTTP, the next-generation transport protocol designed to succeed SSE for MCP communications.

A quick refresher

Before we start, let's make sure we all understand the following terms:

• SSE enables servers to push data to clients over a persistent HTTP connection. In a typical setup for MCP, this involved using two distinct endpoints: one for the client to send requests to the server (usually via HTTP POST) and a separate endpoint where the client would establish an SSE connection (HTTP GET) to receive streaming responses and server-initiated messages.

• MCP is an open standard designed to standardize how Large Language Models (LLMs) interact with external data sources, APIs and resources as agent tools, MCP aims to replace the current landscape of fragmented, custom integrations with a universal, standardized framework.

• Streamable HTTP utilizes a single HTTP endpoint for both sending requests from the client to the server, and receiving responses and notifications from the server to the client.

aside_block

<ListValue: [StructValue([('title', '$300 in free credit to try Google Cloud developer tools'), ('body', <wagtail.rich_text.RichText object at 0x3ec05f0d9b80>), ('btn_text', 'Start building for free'), ('href', 'http://console.cloud.google.com/freetrial?redirectPath=/welcome'), ('image', None)])]>

Step 1: Create an MCP server 

You need the following python packages installed in your virtual environment before proceeding. We will be using the uv tool in this blog.

code_block

<ListValue: [StructValue([('code', '"beautifulsoup4==4.12.3",\r\n"google-adk==0.3.0",\r\n"html2text==2024.2.26",\r\n"mcp[cli]==1.5.0",\r\n"requests==2.32.3"'), ('language', 'lang-py'), ('caption', <wagtail.rich_text.RichText object at 0x3ec05f0d9a90>)])]>

Here’s an explanation of the Python code server.py:

• It creates an instance of an MCP server using FastMCP

• It defines a tool called extract_wikipedia_article decorated with @mcp.tool

• It configures an SSE transport mechanism SseServerTransport to enable real-time communication, typically for the MCP server interactions.

• It creates a web application instance using the Starlette framework and defines two routes, message and sse.

• You can read more about SSE transport protocol here.

code_block

<ListValue: [StructValue([('code', '# File server.py\r\n\r\nimport requests\r\nfrom requests.exceptions import RequestException\r\nfrom bs4 import BeautifulSoup\r\nfrom html2text import html2text\r\n\r\nimport uvicorn\r\nfrom starlette.applications import Starlette\r\nfrom starlette.requests import Request\r\nfrom starlette.routing import Route, Mount\r\n\r\nfrom mcp.server.fastmcp import FastMCP\r\nfrom mcp.shared.exceptions import McpError\r\nfrom mcp.types import ErrorData, INTERNAL_ERROR, INVALID_PARAMS\r\nfrom mcp.server.sse import SseServerTransport\r\n\r\n# Create an MCP server instance with an identifier ("wiki")\r\nmcp = FastMCP("wiki")\r\n\r\[email protected]()\r\ndef extract_wikipedia_article(url: str) -> str:\r\n """\r\n Retrieves and processes a Wikipedia article from the given URL, extracting\r\n the main content and converting it to Markdown format.\r\n\r\n Usage:\r\n extract_wikipedia_article("https://en.wikipedia.org/wiki/Gemini_(chatbot)")\r\n """\r\n try:\r\n if not url.startswith("http"):\r\n raise ValueError("URL must begin with http or https protocol.")\r\n\r\n response = requests.get(url, timeout=8)\r\n if response.status_code != 200:\r\n raise McpError(\r\n ErrorData(\r\n code=INTERNAL_ERROR,\r\n message=f"Unable to access the article. Server returned status: {response.status_code}"\r\n )\r\n )\r\n soup = BeautifulSoup(response.text, "html.parser")\r\n content_div = soup.find("div", {"id": "mw-content-text"})\r\n if not content_div:\r\n raise McpError(\r\n ErrorData(\r\n code=INVALID_PARAMS,\r\n message="The main article content section was not found at the specified Wikipedia URL."\r\n )\r\n )\r\n markdown_text = html2text(str(content_div))\r\n return markdown_text\r\n\r\n except Exception as e:\r\n raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"An unexpected error occurred: {str(e)}")) from e\r\n\r\nsse = SseServerTransport("/messages/")\r\n\r\nasync def handle_sse(request: Request) -> None:\r\n _server = mcp._mcp_server\r\n async with sse.connect_sse(\r\n request.scope,\r\n request.receive,\r\n request._send,\r\n ) as (reader, writer):\r\n await _server.run(reader, writer, _server.create_initialization_options())\r\n\r\napp = Starlette(\r\n debug=True,\r\n routes=[\r\n Route("/sse", endpoint=handle_sse),\r\n Mount("/messages/", app=sse.handle_post_message),\r\n ],\r\n)\r\n\r\nif __name__ == "__main__":\r\n uvicorn.run(app, host="localhost", port=8001)'), ('language', 'lang-py'), ('caption', <wagtail.rich_text.RichText object at 0x3ec05f0d9c10>)])]>

To start the server, you can run the command uv run server.py.

Bonus tip, to debug the server using MCP Inspector, execute the command uv run mcp dev server.py.

Step 2: Attach the MCP server while creating ADK agents

The following explains the Python code in the file agent.py:

• Uses MCPToolset.from_server with SseServerParams to establish a SSE connection to a URI endpoint. For this demo we will use http://localhost:8001/sse, but in production this would be a remote server.

• Create an ADK Agent and call get_tools_async to get the tools from the MCP server.

code_block

<ListValue: [StructValue([('code', '# File agent.py\r\n\r\nimport asyncio\r\nimport json\r\nfrom typing import Any\r\n\r\nfrom dotenv import load_dotenv\r\nfrom google.adk.agents.llm_agent import LlmAgent\r\nfrom google.adk.artifacts.in_memory_artifact_service import (\r\n InMemoryArtifactService, # Optional\r\n)\r\nfrom google.adk.runners import Runner\r\nfrom google.adk.sessions import InMemorySessionService\r\nfrom google.adk.tools.mcp_tool.mcp_toolset import (\r\n MCPToolset,\r\n SseServerParams,\r\n)\r\nfrom google.genai import types\r\nfrom rich import print\r\nload_dotenv()\r\n\r\nasync def get_tools_async():\r\n """Gets tools from the File System MCP Server."""\r\n tools, exit_stack = await MCPToolset.from_server(\r\n connection_params=SseServerParams(\r\n url="http://localhost:8001/sse",\r\n )\r\n )\r\n print("MCP Toolset created successfully.")\r\n return tools, exit_stack\r\n\r\nasync def get_agent_async():\r\n """Creates an ADK Agent equipped with tools from the MCP Server."""\r\n tools, exit_stack = await get_tools_async()\r\n print(f"Fetched {len(tools)} tools from MCP server.")\r\n root_agent = LlmAgent(\r\n model="gemini-2.0-flash",\r\n name="assistant",\r\n instruction="""Help user extract and summarize the article from wikipedia link.\r\n Use the following tools to extract wikipedia article:\r\n - extract_wikipedia_article\r\n\r\n Once you retrieve the article, always summarize it in a few sentences for the user.\r\n """,\r\n tools=tools,\r\n )\r\n return root_agent, exit_stack\r\n\r\nroot_agent = get_agent_async()'), ('language', 'lang-py'), ('caption', <wagtail.rich_text.RichText object at 0x3ec05f0d9bb0>)])]>

Step 3: Test your agent

We will use the ADK developer tool to test the agent.

Create the following directory structure:

code_block

<ListValue: [StructValue([('code', '. # <--Your current directory\r\n├── adk-agent\r\n│ ├── __init__.py\r\n│ └── agent.py\r\n├── .env'), ('language', 'lang-py'), ('caption', <wagtail.rich_text.RichText object at 0x3ec05f0d98e0>)])]>

The content for __init__.py and .env are as follows:

code_block

<ListValue: [StructValue([('code', '# .env\r\nGOOGLE_GENAI_USE_VERTEXAI="True"\r\nGOOGLE_CLOUD_PROJECT=<YOUR_PROJECT_ID>\r\nGOOGLE_CLOUD_LOCATION="us-central1"'), ('language', 'lang-py'), ('caption', <wagtail.rich_text.RichText object at 0x3ec05f0d9280>)])]>

code_block

<ListValue: [StructValue([('code', '# __init__.py\r\nfrom . import agent'), ('language', 'lang-py'), ('caption', <wagtail.rich_text.RichText object at 0x3ec05f0d9e20>)])]>

Start the UI with the following command:

code_block

<ListValue: [StructValue([('code', 'uv run adk web'), ('language', 'lang-py'), ('caption', <wagtail.rich_text.RichText object at 0x3ec05f0d9c70>)])]>

This will open up the ADK developer tool interface as shown below:

Streamable HTTP

It is worth noting that in March 2025, MCP released a new transport protocol called Streamable HTTP. The Streamable HTTP transport allows a server to function as an independent process managing multiple client connections via HTTP POST and GET requests. Servers can optionally implement Server-Sent Events (SSE) for streaming multiple messages, enabling support for basic MCP servers as well as more advanced servers with streaming and server-initiated communication.

The following code demonstrates how to implement a Streamable HTTP server, where the tool extract_wikipedia_article will return a dummy string to simplify the code.

code_block

<ListValue: [StructValue([('code', '# File server.py\r\n\r\nimport contextlib\r\nimport logging\r\nfrom collections.abc import AsyncIterator\r\n\r\nimport anyio\r\nimport mcp.types as types\r\nfrom mcp.server.lowlevel import Server\r\nfrom mcp.server.streamable_http_manager import StreamableHTTPSessionManager\r\nfrom starlette.applications import Starlette\r\nfrom starlette.routing import Mount\r\nfrom starlette.types import Receive, Scope, Send\r\nimport uvicorn\r\n\r\nlogger = logging.getLogger(__name__)\r\n\r\n\r\napp = Server("mcp-streamable-http-stateless-demo")\r\n\r\n\r\[email protected]_tool()\r\nasync def call_tool(\r\n name: str, arguments: dict\r\n) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:\r\n # Check if the tool is extract-wikipedia-article\r\n if name == "extract-wikipedia-article":\r\n # Return dummy content for the Wikipedia article\r\n return [\r\n types.TextContent(\r\n type="text",\r\n text="This is the article ...",\r\n )\r\n ]\r\n\r\n # For other tools, keep the existing notification logic\r\n ctx = app.request_context\r\n interval = arguments.get("interval", 1.0)\r\n count = arguments.get("count", 5)\r\n caller = arguments.get("caller", "unknown")\r\n\r\n # Send the specified number of notifications with the given interval\r\n for i in range(count):\r\n await ctx.session.send_log_message(\r\n level="info",\r\n data=f"Notification {i + 1}/{count} from caller: {caller}",\r\n logger="notification_stream",\r\n related_request_id=ctx.request_id,\r\n )\r\n if i < count - 1: # Don\'t wait after the last notification\r\n await anyio.sleep(interval)\r\n\r\n return [\r\n types.TextContent(\r\n type="text",\r\n text=(\r\n f"Sent {count} notifications with {interval}s interval"\r\n f" for caller: {caller}"\r\n ),\r\n )\r\n ]\r\n\r\n\r\[email protected]_tools()\r\nasync def list_tools() -> list[types.Tool]:\r\n return [\r\n types.Tool(\r\n name="extract-wikipedia-article",\r\n description=("Extracts the main content of a Wikipedia article"),\r\n inputSchema={\r\n "type": "object",\r\n "required": ["url"],\r\n "properties": {\r\n "url": {\r\n "type": "string",\r\n "description": "URL of the Wikipedia article to extract",\r\n },\r\n },\r\n },\r\n )\r\n ]\r\n\r\n\r\nsession_manager = StreamableHTTPSessionManager(\r\n app=app,\r\n event_store=None,\r\n stateless=True,\r\n)\r\n\r\n\r\nasync def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None:\r\n await session_manager.handle_request(scope, receive, send)\r\n\r\n\r\[email protected]\r\nasync def lifespan(app: Starlette) -> AsyncIterator[None]:\r\n """Context manager for session manager."""\r\n async with session_manager.run():\r\n logger.info("Application started with StreamableHTTP session manager!")\r\n try:\r\n yield\r\n finally:\r\n logger.info("Application shutting down...")\r\n\r\n\r\napp = Starlette(\r\n debug=True,\r\n routes=[\r\n Mount("/mcp", app=handle_streamable_http),\r\n ],\r\n lifespan=lifespan,\r\n)\r\n\r\nif __name__ == "__main__":\r\n uvicorn.run(app, host="localhost", port=3000)'), ('language', 'lang-py'), ('caption', <wagtail.rich_text.RichText object at 0x3ec05f0cd520>)])]>

You can start the Streamable HTTP MCP server using by running the following:

code_block

<ListValue: [StructValue([('code', '# Start the server\r\nuv run server.py'), ('language', 'lang-py'), ('caption', <wagtail.rich_text.RichText object at 0x3ec05f0cd790>)])]>

To debug with MCP Inspector, select Streamable HTTP and fill in the MCP Server URL http://localhost:3000/mcp.

Authentication

For production deployments of MCP servers, robust authentication is a critical security consideration. As this field is under active development at the time of writing, we recommend referring to the MCP specification on Authentication for more information.

For an enterprise grade API governance system which, similar to MCP, can generate agent tools:

• Apigee centralizes and manages any APIs, with full control, versioning, and governance 

• API Hub organizes metadata for any API and documentation 

• Application Integrations support many existing API connections with user access control support 

• ADK supports these Google Cloud Managed Tools with about the same number of lines of code

Get started 

To get started today, read the documentation for ADK. You can create your own Agent with access to available MCP servers in the open community with ADK.