Cloud RunでMCP Serverを構築してGoogle Chatにメッセージ送信してみた！

はじめに

AI Agentの時代が本格的に到来していますが、開発者の皆さんはこんな悩みを抱えていませんか？

「Gemini、Claudeなどの優秀なAIモデルがあるのに、自社のデータベースやAPIと連携させるのが面倒…」 「カスタムツールを作っても、複数のAIプラットフォームで使い回せない…」 「セキュアで本格的な外部連携を実装するには、認証やらインフラやら考えることが多すぎる…」

そんな課題を一気に解決してくれるのがMCP (Model Context Protocol) です。一度構築すれば、複数のAIプラットフォームから同じツールを利用でき、しかもCloud Runを使えばスケーラブルで運用も楽々。今回は、実際にCloud Run上でMCP Serverを構築し、Google Chatにメッセージを送信する手順を詳しく解説します。

対象読者

• AI AgentやLLMアプリケーションの開発に興味がある方 外部APIやデータベースとAIを連携させたい方
• Google Cloud（特にCloud Run）の基本的な操作ができる方
• Docker コンテナとPythonの基礎知識がある方
• 「AIは使えるけど、本格的なツール連携はまだ…」という方
• Google Chatを業務で使っていて、AIとの連携に興味がある方
• 「AIに通知業務を任せられたら楽になりそう」と思っている方

MCPとは

MCP（Model Context Protocol）は、AIモデル（例: Gemini, Claude）が外部のデータソースやツールと連携するための標準化されたプロトコルです。これにより、LLMが最新の情報や外部サービスを効率的に利用できるようになり、開発の効率化やAIの活用の幅を広げることが期待されています。

なぜMCPが必要なのか？

従来のLLM（大規模言語モデル）は、学習データに基づいてテキストを生成するだけで、外部のデータやツールを直接操作することができませんでした。そのため、例えばチャットにメッセージを送信したり、カレンダーを操作したりするためには、個別のAPIを実装する必要があり、開発の複雑さが増していました。

MCPは、これらの問題を解決するために登場しました。AIが外部のツールやデータにアクセスするための共通のインターフェースを提供することで、開発者はMCPに対応したツールであれば、容易にAIに連携させることができます。

今回作るもの：Google Chat連携MCP Server

本記事では、Google Chatにメッセージを送信するMCP Serverを実際に構築します。

実装する機能

Google Chat Webhook連携

• チャットスペースへのメッセージ送信

MCP Server実装

• Google Chat送信ツールの定義

構築手順

Google Chatのスペースを作成

MCP Serverからの通知用スペースを作成します。

• 「新しいチャット」→「スペースを作成」を押下
  

• 任意のスペース名(今回はmcp-demo)を入力し、「作成」を押下
  

これでスペースが作成できました！ここからはWebhookの設定をしていきます。

• スペース名の右側の「ｖ」を押下後に表示される「アプリと統合」を押下

• 「+ Webhookを追加」を押下

• 任意の名称(今回はmcp-demo)を入力し「保存」を押下
  

• Webhookの設定が完了しました！

※Webhook URLは動作確認の際に必要になるので、控えておいてください！

MCP Server実装

• ディレクトリ構成

mcp-server
├── Dockerfile
├── gws
│   ├── __init__.py
│   ├── clients
│   │   ├── __init__.py
│   │   └── chat.py
│   ├── models
│   │   ├── __init__.py
│   │   └── responses.py
│   └── tools
│       ├── __init__.py
│       └── chat_message.py
├── pyproject.toml
├── server.py
├── tests
│   ├── http_client.py
│   └── model.py
└── uv.lock

• ライブラリインストール

uv add fastmcp httpx pydantic requests

• gws/clinets/chat.py: チャットスペースのWebhookを使用しメッセージを送信するclass

import requests

class WebhookChatClient:
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url

    def send_message(self, text: str):
        try:
            payload = {"text": text}
            response = requests.post(
                self.webhook_url,
                json=payload,
                headers={"Content-Type": "application/json"},
            )

            if response.status_code == 200:
                print("メッセージが正常に送信されました")
                return True
            else:
                print(f"送信エラー: {response.status_code} - {response.text}")
                return False

        except Exception as e:
            print(f"送信エラー: {e}")
            return False

• gws/models/responses.py: MCP Serverが返却するメッセージのmodel

from typing import Optional, Any

from pydantic import BaseModel, Field


class MCPToolResponse(BaseModel):
    """MCP tool response format"""

    success: bool = Field(description="Tool execution success")
    message: Optional[str] = Field(default=None, description="Response message")
    sent_payload: Optional[dict[str, Any]] = Field(
        default=None, description="Request payload sent"
    )
    error: Optional[str] = Field(default=None, description="Error details")

• tools/chat_message.py: WebhookでGWSチャットにメッセージを送信するMCPツール登録関数

from typing import Annotated

from fastmcp import FastMCP
from pydantic import Field

from gws.clients.chat import WebhookChatClient
from gws.models.responses import MCPToolResponse


def register_gws_chat_tools(mcp: FastMCP) -> None:
    @mcp.tool(tags={"public", "gws", "chat"})
    async def send_gws_chat_message(
        webhook_url: Annotated[str, Field(description="送信先WebhookURL")],
        message: Annotated[str, Field(description="送信メッセージ")],
    ) -> MCPToolResponse:
        try:
            client = WebhookChatClient(webhook_url)
            client.send_message(message)
            return MCPToolResponse(
                success=True,
                message="The message was successfully sent.",
                sent_payload={
                    "webhookUrl": webhook_url,
                    "message": message,
                },
            )

        except Exception as e:
            print(f"Exception occurred while sending message: {str(e)}")
            return MCPToolResponse(
                success=False,
                error=f"送信エラー: {str(e)}",
                sent_payload={
                    "webhookUrl": webhook_url,
                    "message": message,
                },
            )

• server.py: MCP Server起動スクリプト

import os
import json

from fastmcp import FastMCP

from gws.tools.chat_message import register_gws_chat_tools


def pydantic_object_serializer(data):
    return json.dumps(data.model_dump(), ensure_ascii=False, indent=2)


def create_server() -> FastMCP:
    mcp = FastMCP(
        name="MCPServer",
        include_tags={"public"},
        on_duplicate_tools="error",
        on_duplicate_resources="warn",
        on_duplicate_prompts="replace",
        tool_serializer=pydantic_object_serializer,
    )

    register_gws_chat_tools(mcp)

    return mcp


def main() -> None:
    print("Starting Message Sender MCP Server")

    mcp = create_server()
    mcp.run(
        transport="streamable-http",
        host=os.getenv("HOST", "0.0.0.0"),
        port=int(os.getenv("PORT", "9001")),
    )

if __name__ == "__main__":
    main()

Cloud Runへデプロイ

• ここからは、Google CloudのCloud RunにMCP Serverをデプロイしていきます。先ずは、MCP ServerのDocker Imageを格納するArtifact Registoryのリポジトリを作成していきます。

• 下記情報を入力&設定し、作成ボタンを押下
  名前: 任意の名称
  形式: Docker
  リージョン: 任意のリージョン

• MCP ServerのDocker Imageを保管するリポジトリが完成しました！

MCP ServerのDocker Imageをbuild

• Dockefile

# Build stage
FROM python:3.13-slim-bookworm as builder

WORKDIR /build
COPY pyproject.toml uv.lock .python-version ./

# Install build dependencies
RUN apt-get update && apt-get upgrade -y && \
    apt-get install -y --no-install-recommends \
    gcc \
    libc6-dev \
    pkg-config \
    && apt-get clean \
    && rm -rf /var/lib/apt/lists/*

# Install uv and generate requirements.txt
RUN pip install --no-cache-dir uv && \
    uv pip compile pyproject.toml -o requirements.txt

# Install dependencies to be copied to the final image
RUN pip install --no-cache-dir --no-warn-script-location --user -r requirements.txt

# Final stage
FROM python:3.13-slim-bookworm

# Set environment variables
ENV PYTHONUNBUFFERED=True
ENV APP_HOME=/app
WORKDIR $APP_HOME

# Copy Python packages and scripts from builder
COPY --from=builder /root/.local/lib/python3.13/site-packages /usr/local/lib/python3.13/site-packages
COPY --from=builder /root/.local/bin /usr/local/bin

# ソースコードコピー
COPY gws/ ./gws/
COPY server.py ./

EXPOSE 8080

CMD ["python", "server.py"]

• gcloudを使用してimage build

gcloud builds submit --region=asia-northeast1 --tag asia-northeast1-docker.pkg.dev/[ProjectID]/mcp-demo/mcp-server:0.0.1

• コマンドが正常終了したことを確認します。

• 先ほど作成したArtifact Registoryのリポジトリにもしっかり登録されています！

Cloud Runにデプロイ

• gcloudを使用し、Artifact Registoryに登録したDocker ImageをCloud Runにデプロイします。

gcloud run deploy mcp-server --image asia-northeast1-docker.pkg.dev/[Project ID]/mcp-demo/mcp-server:0.0.1 --region asia-northeast1 --platform managed --allow-unauthenticated

• Cloud RunにMCP Serverをデプロイできました。
  ※今回は検証のため、「未認証を許可」でデプロイしましたが、本番では認証が必要な設定でデプロイしましょう！

動作確認

• tests/http_clinet.py: MCP Serverにメッセージを送信するツール

import asyncio
from fastmcp import Client


async def test_http_mcp_tool():
    """Streamable HTTP MCPサーバー経由でツールをテスト"""

    # パラメータ
    webhook_url = "[チャットスペースに追加したWebhook URL]"
    message = "hello mcp"

    print("HTTP MCPクライアントでGWSメッセージ送信テストを開始...")
    print(f"Message: {message}")
    print("-" * 50)

    client = Client("[デプロイしたCloud RunのURL]/mcp/")
    # client = Client("http://127.0.0.1:9001/mcp")
    try:
        async with client:
            # MCPサーバで提供されているツール一覧の取得
            tools = await client.list_tools()
            print(f"Available tools: {tools}")

            # ツール呼び出し
            results = await client.call_tool(
                "send_gws_chat_message",
                {"webhook_url": webhook_url, "message": message},
            )

            for result in results:
                print("---")
                print(result)

    except Exception as e:
        print(f"fastmcp.Client エラー: {e}")
        print(f"エラータイプ: {type(e)}")


if __name__ == "__main__":
    asyncio.run(test_http_mcp_tool())

• メッセージ送信ツール実行

 python tests/http_client.py

• 結果確認

## ツール実行結果
type='text' 
text='{
  "success": true,
  "message": "The message was successfully sent.",
  "sent_payload": {
     "webhookUrl": "[実際のWebhook URL]", 
     "message": "hello mcp"  
  },
  "error": null
}' 
annotations=None

• Cloud Run(MCP Sever)ログ

• Google Chatスペース

Google Chatのスペースにメッセージが投稿されました！
Cloud RunにデプロイしたMCP Serverも動作していることが確認できましたね！

まとめ

Cloud Run上でのMCP Server構築は、AI時代の外部連携課題を解決する強力なソリューションです。

今回の構築で得られるメリット

• 複数AIプラットフォーム対応の統一ツール
• スケーラブルなインフラ
• 運用コストの最小化
• 将来的な拡張性の確保

今後検討が必要な課題

• ClientとCloud Run間の認証
• Cloud RunとGoogle Chat認証

MCPはまだ発展途上の技術ですが、その可能性は計り知れません。今のうちに基礎を固めておくことで、AIファーストな開発体制の構築において大きなアドバンテージを得られるでしょう。

皆さんもぜひ、この機会にMCPの世界に飛び込んでみてください！