デベロッパー

Google ADK と MCP の外部サーバーとのインテグレーションガイド

2025年5月28日

Wei Yih Yap

Generative AI Field Solutions Architect

Alan Blount

Product Manager

Try Gemini 2.5

Our most intelligent model is now available on Vertex AI

Try now

※この投稿は米国時間 2025 年 5 月 15 日に、Google Cloud blog に投稿されたものの抄訳です。

AI を活用したエージェントが有益かつ実用的なタスクを実行するには、ベースモデルの外部にあるツールや最新情報に確実にアクセスできる必要があります。Anthropic の Model Context Protocol（MCP）は、この点に対処するために設計されており、エージェントが応答や対応を通知するために必要とする重要な外部コンテキストを取得するための、標準化された方法を提供するものです。

これは企業データや一般公開ツールの活用が可能な高度なエージェントの構築とデプロイを必要とするデベロッパーにとっては不可欠です。しかし、Google のエージェント開発キット（ADK）で構築されたエージェントを統合して、MCP サーバー、特に外部でホストされているサーバーと効果的に通信させようとすると、インテグレーションに関する問題がいくつか生じる可能性があります。

今回は、外部の MCP サーバーに接続する ADK エージェントの開発について、まずは、サーバー送信イベント（SSE）の使用について説明します。MCP を活用して Wikipedia の記事にアクセスする ADK エージェントの例を紹介します。これは、外部の専門的なデータを取得する一般的なユースケースです。また、MCP 通信における SSE の後継として設計された次世代トランスポートプロトコルである Streamable HTTP も紹介します。

簡単な復習

始める前に、以下の用語を理解していることを確認します。

SSE : SSE を使用すると、サーバーは永続的な HTTP 接続を介してクライアントにデータを push できます。MCP の一般的な設定では、2 つの異なるエンドポイントを使用します。1 つは、クライアントがサーバーにリクエストを送信するためのエンドポイント（通常は HTTP POST 経由）で、もう 1 つは、クライアントが SSE 接続を確立して（HTTP GET）、ストリーミング応答とサーバー起動のメッセージを受信するための個別のエンドポイントです。
MCP: 大規模言語モデル（LLM）がエージェントツールとして外部のデータソース、API、リソースとやり取りする方法を標準化するために設計されたオープン標準です。断片化されたカスタムインテグレーションの現在の状況を、普遍的で標準化されたフレームワークに置き換えることを目指しています。
Streamable HTTP: クライアントからサーバーへのリクエストの送信と、サーバーからクライアントへのレスポンスと通知の受信の両方に単一の HTTP エンドポイントを使用します。

ステップ 1: MCP サーバーを作成する

次に進む前に、次の Python パッケージを仮想環境にインストールしておく必要があります。このブログ記事では、uv ツールを使用します。

lang-py

読み込んでいます...

Python コード server.py についての説明は次のとおりです。

FastMCP を使用して MCP サーバーのインスタンスを作成します。
@mcp.tool で修飾された extract_wikipedia_article というツールを定義します。
MCP サーバーとのやり取りに一般的に使用される、リアルタイム通信を可能にする SSE トランスポートメカニズム SseServerTransport を構成します。
Starlette フレームワークを使用してウェブアプリケーションインスタンスを作成し、message と sse の 2 つのルートを定義します。
SSE トランスポートプロトコルの詳細については、こちらをご覧ください。

lang-py

読み込んでいます...

# File server.py

import requests
from requests.exceptions import RequestException
from bs4 import BeautifulSoup
from html2text import html2text

import uvicorn
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.routing import Route, Mount

from mcp.server.fastmcp import FastMCP
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData, INTERNAL_ERROR, INVALID_PARAMS
from mcp.server.sse import SseServerTransport

# Create an MCP server instance with an identifier ("wiki")
mcp = FastMCP("wiki")

@mcp.tool()
def extract_wikipedia_article(url: str) -> str:
    """
    Retrieves and processes a Wikipedia article from the given URL, extracting
    the main content and converting it to Markdown format.

Usage:
        extract_wikipedia_article("https://en.wikipedia.org/wiki/Gemini_(chatbot)")
    """
    try:
        if not url.startswith("http"):
            raise ValueError("URL must begin with http or https protocol.")

response = requests.get(url, timeout=8)
        if response.status_code != 200:
            raise McpError(
                ErrorData(
                    code=INTERNAL_ERROR,
                    message=f"Unable to access the article. Server returned status: {response.status_code}"
                )
            )
        soup = BeautifulSoup(response.text, "html.parser")
        content_div = soup.find("div", {"id": "mw-content-text"})
        if not content_div:
            raise McpError(
                ErrorData(
                    code=INVALID_PARAMS,
                    message="The main article content section was not found at the specified Wikipedia URL."
                )
            )
        markdown_text = html2text(str(content_div))
        return markdown_text

except Exception as e:
        raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"An unexpected error occurred: {str(e)}")) from e

sse = SseServerTransport("/messages/")

async def handle_sse(request: Request) -> None:
    _server = mcp._mcp_server
    async with sse.connect_sse(
        request.scope,
        request.receive,
        request._send,
    ) as (reader, writer):
        await _server.run(reader, writer, _server.create_initialization_options())

app = Starlette(
    debug=True,
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ],
)

if __name__ == "__main__":
    uvicorn.run(app, host="localhost", port=8001)

サーバーを起動するには、コマンド uv run server.py を実行します。

追加のヒント: MCP Inspector を使用してサーバーをデバッグするには、コマンド uv run mcp dev server.py を実行します。

https://storage.googleapis.com/gweb-cloudblog-publish/images/01-mcp-inspector.max-1800x1800.png

ステップ 2: ADK エージェントの作成時に MCP サーバーをアタッチする

次に、agent.py ファイルの Python コードについて説明します。

MCPToolset.from_server と SseServerParams を使用して、URI エンドポイントへの SSE 接続を確立します。このデモでは http://localhost:8001/sse を使用しますが、本番環境ではリモートサーバーを使用します。
ADK エージェントを作成し、get_tools_async を呼び出して MCP サーバーからツールを取得します。

lang-py

読み込んでいます...

# File agent.py

import asyncio
import json
from typing import Any

from dotenv import load_dotenv
from google.adk.agents.llm_agent import LlmAgent
from google.adk.artifacts.in_memory_artifact_service import (
    InMemoryArtifactService,  # Optional
)
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.mcp_tool.mcp_toolset import (
    MCPToolset,
    SseServerParams,
)
from google.genai import types
from rich import print
load_dotenv()

async def get_tools_async():
    """Gets tools from the File System MCP Server."""
    tools, exit_stack = await MCPToolset.from_server(
        connection_params=SseServerParams(
            url="http://localhost:8001/sse",
        )
    )
    print("MCP Toolset created successfully.")
    return tools, exit_stack

async def get_agent_async():
    """Creates an ADK Agent equipped with tools from the MCP Server."""
    tools, exit_stack = await get_tools_async()
    print(f"Fetched {len(tools)} tools from MCP server.")
    root_agent = LlmAgent(
        model="gemini-2.0-flash",
        name="assistant",
        instruction="""Help user extract and summarize the article from wikipedia link.
        Use the following tools to extract wikipedia article:
        - extract_wikipedia_article

Once you retrieve the article, always summarize it in a few sentences for the user.
        """,
        tools=tools,
    )
    return root_agent, exit_stack

root_agent = get_agent_async()

ステップ 3: エージェントをテストする

ADK デベロッパーツールを使用してエージェントをテストします。

次のディレクトリ構造を作成します。

lang-py

読み込んでいます...

__init__.py と .env のコンテンツは次のとおりです。

lang-py

読み込んでいます...

lang-py

読み込んでいます...

次のコマンドで UI を起動します。

lang-py

読み込んでいます...

次に示すように、ADK デベロッパーツールのインターフェースが開きます。

https://storage.googleapis.com/gweb-cloudblog-publish/images/02-adk-web.max-1800x1800.png

Streamable HTTP

2025 年 3 月に MCP は、 Streamable HTTP という新しいトランスポートプロトコルをリリースしました。Streamable HTTP トランスポートを使用すると、サーバーが HTTP POST リクエストと GET リクエストを介して複数のクライアント接続を管理する独立したプロセスとして機能します。サーバーには、複数のメッセージをストリーミングするためのサーバー送信イベント（SSE）をオプションで実装できます。実装すると、基本的な MCP サーバーだけでなく、ストリーミングや、サーバーによって開始される通信を備えたより高度なサーバーのサポートも可能となります。

次のコードは、Streamable HTTP サーバーの実装方法を示しています。ここでは、コードを簡素化するために、extract_wikipedia_article ツールはダミー文字列を返します。

lang-py

読み込んでいます...

# File server.py

import contextlib
import logging
from collections.abc import AsyncIterator

import anyio
import mcp.types as types
from mcp.server.lowlevel import Server
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.routing import Mount
from starlette.types import Receive, Scope, Send
import uvicorn

logger = logging.getLogger(__name__)

app = Server("mcp-streamable-http-stateless-demo")

@app.call_tool()
async def call_tool(
    name: str, arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    # Check if the tool is extract-wikipedia-article
    if name == "extract-wikipedia-article":
        # Return dummy content for the Wikipedia article
        return [
            types.TextContent(
                type="text",
                text="This is the article ...",
            )
        ]

# For other tools, keep the existing notification logic
    ctx = app.request_context
    interval = arguments.get("interval", 1.0)
    count = arguments.get("count", 5)
    caller = arguments.get("caller", "unknown")

# Send the specified number of notifications with the given interval
    for i in range(count):
        await ctx.session.send_log_message(
            level="info",
            data=f"Notification {i + 1}/{count} from caller: {caller}",
            logger="notification_stream",
            related_request_id=ctx.request_id,
        )
        if i < count - 1:  # Don't wait after the last notification
            await anyio.sleep(interval)

return [
        types.TextContent(
            type="text",
            text=(
                f"Sent {count} notifications with {interval}s interval"
                f" for caller: {caller}"
            ),
        )
    ]

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="extract-wikipedia-article",
            description=("Extracts the main content of a Wikipedia article"),
            inputSchema={
                "type": "object",
                "required": ["url"],
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "URL of the Wikipedia article to extract",
                    },
                },
            },
        )
    ]

session_manager = StreamableHTTPSessionManager(
    app=app,
    event_store=None,
    stateless=True,
)

async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None:
    await session_manager.handle_request(scope, receive, send)

@contextlib.asynccontextmanager
async def lifespan(app: Starlette) -> AsyncIterator[None]:
    """Context manager for session manager."""
    async with session_manager.run():
        logger.info("Application started with StreamableHTTP session manager!")
        try:
            yield
        finally:
            logger.info("Application shutting down...")

app = Starlette(
    debug=True,
    routes=[
        Mount("/mcp", app=handle_streamable_http),
    ],
    lifespan=lifespan,
)

if __name__ == "__main__":
    uvicorn.run(app, host="localhost", port=3000)

Streamable HTTP MCP サーバーを起動するには、次のコマンドを実行します。

lang-py

読み込んでいます...

MCP Inspector でデバッグするには、[Streamable HTTP] を選択し、MCP サーバー URL http://localhost:3000/mcp を入力します。

https://storage.googleapis.com/gweb-cloudblog-publish/images/03-streamable-http.max-2100x2100.png

認証

MCP サーバーを本番環境にデプロイする場合、堅牢な認証は重要なセキュリティ上の考慮事項です。この分野は執筆時点で活発に開発が進められており、詳細については認証に関する MCP の仕様を参照することをおすすめします。

MCP と同様にエージェントツールを生成できるエンタープライズグレードの API ガバナンスシステムの場合:

Apigee は、あらゆる API を一元管理し、完全な制御、バージョニング、ガバナンスを実現
API ハブは、あらゆる API とドキュメントのメタデータを整理
Application Integration は、ユーザーアクセス制御をサポートする多くの既存の API 接続をサポート
ADK は、これらの Google Cloud マネージドツールを、ほぼ同じコード行数でサポート

使ってみる

今すぐ利用を開始するには、ADK のドキュメントをご覧ください。ADK を使用して、オープンコミュニティで利用可能な MCP サーバーにアクセスできる独自のエージェントを作成できます。

-生成 AI ソリューションアーキテクト、Wei Yih Yap

投稿先

https://storage.googleapis.com/gweb-cloudblog-publish/images/hero_-_cloud_storage.max-700x700.png

Developers & Practitioners

スケーラブルな AI はストレージから: モデルアーティファクト戦略ガイド

執筆者: Karl Weinmeister • 所要時間: 11 分

Containers & Kubernetes

IP アドレス管理がシンプルに: GKE クラスタの自動 IPAM を発表

執筆者: Whitney Jenkins • 所要時間: 2 分

DevOps & SRE

Google Cloud の設定のご紹介: 安全なクラウド基盤へのガイド付きパスウェイ

執筆者: Stacie Buckingham • 所要時間: 4 分

https://storage.googleapis.com/gweb-cloudblog-publish/images/State_and_Memory_hero_image_v2.max-700x700.png

Developers & Practitioners

ADK を使用したエージェントの状態と記憶

執筆者: Megan O'Keefe • 所要時間: 7 分

Google ADK と MCP の外部サーバーとのインテグレーション ガイド

Wei Yih Yap

Alan Blount

Try Gemini 2.5

関連記事

スケーラブルな AI はストレージから: モデル アーティファクト戦略ガイド

IP アドレス管理がシンプルに: GKE クラスタの自動 IPAM を発表

Google Cloud の設定のご紹介: 安全なクラウド基盤へのガイド付きパスウェイ

ADK を使用したエージェントの状態と記憶

Google ADK と MCP の外部サーバーとのインテグレーションガイド

スケーラブルな AI はストレージから: モデルアーティファクト戦略ガイド