Gemini Enterprise AIエージェントの問い合わせ受付を共通化するMCPサーバーを作った

こんにちは、 mhosoya です

弊社では、 Gemini Enterprise を活用した AIエージェント の開発に取り組んでいます

AIエージェント は、公開して終わりではなく、 継続的な改善が必要 だと考えています
つまり、 運用が大事 ということです

開発段階ですべてのあらゆる想定をすることは難しいです
そのため、実際に使ってもらって、「期待した回答が得られない」「こういう機能があると便利」といった課題や要望から改善に繋げるのが一番現実的だと思います

また、こうしたユーザーの意見は、 改善につながりやすい ので、非常に貴重です！

なので、今回は AIエージェント との会話の中から利用できる 問い合わせ機能 を用意したいと思います
そうすれば、何か困ったときに確認先を探したり、別のフォームを開いたりせず、その場で不具合や要望を伝えられるので良いと思います

（理想で言うと、下記のサイクルを回したい）

また、問い合わせ機能については、 AIエージェント ごとに実装しても機能はほぼ変わらないはずなので、 汎用的 に作ってみたいと思います

なので、この機能を複数の AIエージェント で利用できるように、 MCPサーバー で作っていくことにします

各 AIエージェント はユーザーとの会話と送信前の確認を担当し、 MCPサーバー が問い合わせ元に応じた保存先へ登録をします

本記事では、まず接続確認用の最小構成で MCPサーバー を作成し、 Cloud Run へのデプロイと AIエージェント からの呼び出しを確認します
その後、問い合わせ受付を実装して、複数の AIエージェント で共通で使用する仕組みを紹介したいと思います

構成

ADK製 の AIエージェント と Workflow Builder(旧Agent Designer)製 の AIエージェント の両方から接続できるように目指してみたいと思います

ここで、方針について決めておきたいと思います

導入コストが高いと、そもそも組み込まないので、ここは一番気にしたいところです

今回作る問い合わせMCPサーバーには、何で作ったかで 接続方式 が変わります

• ADK製
  • 直接接続
• Workflow Builder(旧Agent Designer)製
  • データストア経由で接続

構築

MCPサーバー

Python で MCPサーバー を実装していきます
まずは、 最小構成 でつくってみます

uv でつくっていきます

uv python install 3.14

uv init --package --python 3.14 inquiry-mcp
cd inquiry-mcp

uv add "mcp==2.2.0"

最小構成 で動かすための実装をする

"""接続確認用ツールを公開するMCPサーバー。"""

import os

from mcp.server import MCPServer

mcp = MCPServer("inquiry-mcp")


@mcp.tool()
def check_connection() -> dict[str, str]:
    """問い合わせを保存せず、MCP接続の成功を返す。"""
    return {
        "status": "ok",
        "message": "SUCCEEDED!",
    }


def main() -> None:
    """環境変数PORTで指定されたポートでサーバーを起動する。"""
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=int(os.environ.get("PORT", "8080")),
        stateless_http=True,
        json_response=True,
    )


if __name__ == "__main__":
    main()

FROM cgr.dev/chainguard/python:latest-dev@sha256:073219a510343c070f74fd1fd7100b9e599cf68eb1536be8ebd4c0b170ede589 AS build

ENV UV_PYTHON_DOWNLOADS=never \
    UV_NO_CACHE=1 \
    UV_PROJECT_ENVIRONMENT=/home/nonroot/.venv

WORKDIR /app

COPY pyproject.toml uv.lock .python-version README.md ./
COPY src/ ./src/

RUN uv sync --locked --no-dev --no-editable

USER root
RUN chown -R 0:0 /home/nonroot/.venv \
    && chmod -R a-w /home/nonroot/.venv


FROM cgr.dev/chainguard/python:latest@sha256:3402da0629d26501855f13d490b7fa1b525e4a5db3a10af324277e5507fcb8e6 AS runtime

ENV PATH="/home/nonroot/.venv/bin:$PATH" \
    PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PORT=8080

WORKDIR /app

COPY --from=build /home/nonroot/.venv /home/nonroot/.venv

USER 65532:65532

EXPOSE 8080

ENTRYPOINT ["/home/nonroot/.venv/bin/inquiry-mcp"]

ローカルで起動してみる

uv run inquiry-mcp

確認する

open http://127.0.0.1:8000/mcp

下記のような感じ

: ping - 2026-09-18 03:42:57.632205+00:00

サーバーからの hearbeat が確認できるはずです

Cloud Run にデプロイする

Cloud Run にデプロイをしていきます

PROJECT_ID="your-project-id"
REGION="asia-northeast1"
REPOSITORY="your-repository"
SERVICE_NAME="inquiry-mcp"
SERVICE_ACCOUNT_NAME="your-service-account"
IMAGE_TAG="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPOSITORY}/${SERVICE_NAME}:v0.1.0"

docker build して、コンテナイメージを作成します

docker buildx build \
--platform linux/amd64 \
--load \
--tag $IMAGE_TAG \
.

Docker から Artifact Registry へ PUSH できるよう、認証を設定します
（これは初回のみ必要）

gcloud auth configure-docker "${REGION}-docker.pkg.dev"

Artifact Registry に PUSH します

docker push $IMAGE_TAG

Cloud Run にデプロイします

gcloud run deploy "$SERVICE_NAME" \
  --project "$PROJECT_ID" \
  --region "$REGION" \
  --image "$IMAGE_TAG" \
  --port 8080 \
  --no-allow-unauthenticated

デプロイした Cloud Run の URL を取得します

gcloud run services describe $SERVICE_NAME \
  --project $PROJECT_ID \
  --region "$REGION" \
  --format json | jq -r '.status.url'

無事に Cloud Run にデプロイできました

Gemini Enterprise の サービスエージェント へ roles/run.invoker を付与します

gcloud run services add-iam-policy-binding inquiry-mcp \
  --project=${PROJECT_ID} \
  --region=asia-northeast1 \
  --member="xxxxx@gcp-sa-discoveryengine.iam.gserviceaccount.com" \
  --role="roles/run.invoker"

Datastore を作成する

下記の設定で Datastore を作成します

• AI Applications > ソース > Custom MCP Server を選択する
• データ > 認証の設定 で下記を設定する
  • 認証方法の選択: No authentication
  • MCP Server URL: （上記で取得した Cloud Run の URL に /mcp をつける）
    e.g. https://xxxxx.asia-northeast1.run.app/mcp
• 構成 > データコネクタの構成
  • データコネクタのロケーション: global
  • データコネクタ名: inquiry-mcp(任意の値)

Datastore が作成できたら、呼び出し可能にする関数を設定します

先ほど、作成したデータストアで下記を実行します

• 操作 > カスタムアクション を再読み込み
• Check Connection を有効にする

最小構成を ADKエージェント 側に組み込んでみる

既存の ADKエージェント に組み込んでみたいと思います

MCPサーバー の Cloud Run は、IAM認証を有効にしたので、 Google 署名付き IDトークン を取得します

下記のような処理を、既存の ADKエージェント に組み込んでみます

import asyncio
import os

from google.adk.tools.mcp_tool.mcp_session_manager import (
    StreamableHTTPConnectionParams,
)
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.auth.transport.requests import Request
from google.oauth2 import id_token

MCP_SERVER_URL = os.environ["MCP_SERVER_URL"].rstrip("/")


def create_auth_headers() -> dict[str, str]:
    """接続先Cloud Run向けのIDトークンを取得する。"""
    token = id_token.fetch_id_token(
        Request(),
        MCP_SERVER_URL,
    )
    return {"Authorization": f"Bearer {token}"}


async def provide_auth_headers(_context: object) -> dict[str, str]:
    """イベントループを止めずに認証ヘッダーを生成する。"""
    return await asyncio.to_thread(create_auth_headers)


mcp_tools = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url=f"{MCP_SERVER_URL}/mcp",
    ),
    header_provider=provide_auth_headers,
    tool_filter=["check_connection"],
)

Instruction には下記のような感じで指定してみます

接続確認を依頼されたらcheck_connectionを実行してください。
ツールの応答に基づいて結果を報告してください。
実行に失敗した場合は成功と報告しないでください。

これで組み込むことはできたので、デプロイしていきます

デプロイが完了したら、対象のエージェントから下記のプロンプトを実行します

check_connection ツールを実行して、MCPサーバーへの接続を確認してください。
返された status と message をそのまま教えてください。

接続はできていそうなので、中身の実装をしていきます

問い合わせ受付で実現したいこと

問い合わせ受付で必要な情報を精査していきます

そのためにまず目的について、改めて考えます

目的 は下記と定義します

ではこの目的の場合に、必要になる情報をまとめていきたいと思います

これにプラスで、調査に必要な情報もまとめます

調査に必要な情報としては、どのエージェントかはもちろんですが、どの Version で起きていたかは重要です

運用していて、問い合わせが来た時に、ちょうど反映したということもあったりします

この時、考えるのは修正内容が足りていなかったかもなどと考えたりします
しかし、 Version を知ることができれば、修正した Version でダメなのか・古い Version でダメだったのかが明確になり、無駄な調査が減ります

そのため、 Version については必須でほしいところですが、各エージェント側に依存するため 任意 にしたいと思います

また、本来であれば スクショ が非常に欲しいところです

しかし、共通問い合わせMCP にしてしまったので、セキュリティの観点からどこに保存してどこまで公開すべきかを考えるのに少し時間がかかるため、今回は一旦外したいと思います

本来であれば、任意で送れるようにしたいですが、今回は間に合わないのでスキップで。。。

複数のエージェントで受付を共通化する

次に複数のエージェントから利用できる構成を考えたいと思います

各エージェントに問い合わせ受付を実装することもできますが、保存処理まで個別に持たせると、項目の追加や保存方法の変更が発生するたびに、各エージェントを修正する必要があります

そこで、問い合わせの検証・保存を 共通のMCPサーバー にまとめ、各エージェントから submit_inquiry ツールを呼び出す構成にします。

役割としては、下記とします

ここからは、接続確認用の check_connection に加えて、問い合わせを受け付ける submit_inquiry を実装していきます

このツールでは、エージェントが聞き取った問い合わせ内容を受け取り、保存先を判定して スプレッドシート へ追記します

処理の流れは次のとおりです。

1. 問い合わせ内容 と 連絡先メールアドレス 、 任意の期待した挙動 を受け取る
2. ヘッダーなどから Agent Id と Version を取得する
3. 必須項目や入力内容を 検証 する
4. エージェントの識別情報を使って、問い合わせ保存先マスタから 保存先 を取得する
5. 受付ID と 受付日時 を付けて保存する
6. 保存結果 をエージェントへ返す

ユーザーから聞き取る情報はツールの引数に、エージェントの識別情報はヘッダーなどに分けて渡します。
保存先はMCPサーバー側で管理し、 LLMに選ばせない 構成にします

保存先のスプレッドシートには、次の情報を登録します

問い合わせの保存先とエージェントの対応を設定

問い合わせ内容が決まったので、次に保存先について考えてみます

今回は、担当者が扱いやすい スプレッドシート を使用することにします
複数のエージェントから共通のMCPサーバーを呼び出しますが、問い合わせの保存先はエージェントごとに設定できるようにしたいと思います

エージェントごとで問い合わせ内容を閲覧できる人を変えたいので、
エージェントが問い合わせ内容の保存先を取得するためのマスタを用意します

スプレッドシートを用意する

スプレッドシートは下記の2つを用意します

• MCPサーバーが問い合わせ内容の保存先を判定するためのマスタ
  • Agent Id と、保存先 のスプレッドシートの対応を管理する
• 問い合わせ内容
  • 受け付けた問い合わせを1件ずつ記録する

それぞれのスプレッドシートには、 Cloud Run で動いている MCPサーバー の サービスアカウント を共有で追加しておきます

MCPサーバーは、呼び出し時にヘッダーに指定された X-Agent-ID を取得します

そして、この X-Agent-ID をもとにどのスプレッドシートに保存するかが決まります

本来なら、ここの部分は厳密な方法で行っても良いですが、 導入コストを低くしたい のでこのままで進めます

問い合わせ受付ツールを実装する

ここまでで準備ができたので、問い合わせを受け付ける submit_inquiry を実装していきます

ここからは、実装のポイントになる部分だけを抜粋して紹介したいと思います
諸々の全体の実装は、別の処理に分けています

まず、エージェントからツールへ渡す情報は、下記のようなのを想定しています

{
  "inquiry_body": "回答に根拠元のリンクが表示されず、確認ができませんでした。",
  "contact_email": "user@example.com",
  "expected_behavior": "回答と一緒に根拠元のリンクも表示してほしいです。"
}

ちなみに agent_id は、この引数には含めません
ユーザーから受け取った情報 と 開発者が設定する情報 を 混在させない ためです

問い合わせ元を取得する

MCPサーバーでは、HTTPリクエストから Agent Id と Version を取得します

取得部分を簡略化すると、次のような感じです
request はMCPのリクエストコンテキストから取得したHTTPリクエストです https://py.sdk.modelcontextprotocol.io/handlers/context/

agent_id = request.headers.get("X-Agent-ID")
if agent_id is None:
    agent_id = request.query_params.get("agent_id")

version = request.headers.get("X-Agent-Version")
if version is None:
    version = request.query_params.get("version")

ヘッダーが省略されている場合だけ、 URL のクエリパラメーターから取得します

これは、 Workflow Builder を考慮してのものになります

Workflow Builder 製の場合は、ヘッダーに設定ができないため、データストアに MCPサーバーの URL を登録する際に、 Agent Id と Version を付加するようにします
（正直、仕方なしという感じでこうしています。。。）

取得した agent_id が空の場合は、保存先を取得できないためエラーにします
また、マスタに登録されていない場合や、受付が無効になっている場合も保存しません

問い合わせをスプレッドシートへ保存する

入力内容と保存先を確認できたら、 受付ID と 受付日時 を付けて保存します

Google Sheets APIへの追記部分は、次のような処理になります
service は認証済みの Sheets API クライアント、保存先と row は検証済みの値を使います

row = [
    reception_id,
    received_at,
    agent_id,
    inquiry_body,
    contact_email,
    expected_behavior if expected_behavior is not None else "",
    version if version is not None else "",
]

escaped_sheet_name = sheet_name.replace("'", "''")

response = (
    service.spreadsheets()
    .values()
    .append(
        spreadsheetId=spreadsheet_id,
        range=f"'{escaped_sheet_name}'!A:G",
        valueInputOption="RAW",
        insertDataOption="OVERWRITE",
        body={"majorDimension": "ROWS", "values": [row]},
    )
    .execute(num_retries=0)
)

valueInputOption="RAW" は、問い合わせ内容を文字列のまま保存するための指定です
これは例えば、本文が = で始まっていても、数式として解釈されないです

また、append で既存の表の末尾へ追記し、insertDataOption="OVERWRITE" で物理的な行挿入を避けています

この呼び出しが返っただけで受付完了とはせず、応答に含まれる保存先や更新行数を確認してから、保存成功として扱っています

保存結果をエージェントへ返す

保存成功を確認できたら、次のような結果を返します

{
  "status": "saved",
  "message": "問い合わせの受付が完了しました。",
  "reception_id": "5bcf0312-4ac8-4f63-8dda-60c0801d3e21",
  "received_at": "2026-09-24T03:00:00.000Z"
}

一方、保存できたか判断できない場合もあります 例えば、スプレッドシートへの追記は完了していても、その応答を受け取る前に通信が切れた場合です

そこで、保存結果は3種類に分けました

unknown のときに自動で再送すると、問い合わせが重複する可能性があります
そのため、保存処理の自動再送を抑止し、エージェントにも保存先を確認する前の再送をさせない方針にしています

なお、コード中の num_retries=0 はAPIクライアントの再試行を止める指定です

ADKエージェントに問い合わせ受付を追加する

接続確認に使ったADKエージェントにも、問い合わせ受付を組み込みます

まず、認証ヘッダーを生成する処理に、 Agent Id と Version を追加します

| Agent Id | 環境変数で指定する | | Version | ADKエージェントの pyproject.toml で管理している値 |

例えば、ADKエージェント側の設定が次の場合です。

[project]
name = "helpdesk-agent"
version = "1.0.0"

アプリをパッケージとしてインストールしていれば、 importlib.metadata.version() で Version を取得できるので、実行時に pyproject.toml を直接読む必要はありません

前半の create_auth_headers() はこうなりました

from importlib.metadata import version

AGENT_ID = os.environ["INQUIRY_AGENT_ID"]
AGENT_VERSION = version("helpdesk-agent")


def create_auth_headers() -> dict[str, str]:
    """IAM認証情報と問い合わせ元のエージェント情報を返す。"""
    token = id_token.fetch_id_token(Request(), MCP_SERVER_URL)

    return {
        "Authorization": f"Bearer {token}",
        "X-Agent-ID": AGENT_ID,
        "X-Agent-Version": AGENT_VERSION,
    }

これは、ADKエージェント自体をインストールする構成が前提となります --no-package の構成ではパッケージ情報が登録されないので、そのままでは取得できないです

ADK側の環境変数には、例えば以下を設定します

INQUIRY_AGENT_ID=helpdesk_agent

続いて、利用するツールに submit_inquiry を追加します。認証ヘッダーを非同期で生成する provide_auth_headers は、前半の実装をそのまま使います

mcp_tools = McpToolset(
    connection_params=StreamableHTTPConnectionParams(
        url=f"{MCP_SERVER_URL}/mcp",
    ),
    header_provider=provide_auth_headers,
    tool_filter=["check_connection", "submit_inquiry"],
)

この mcp_tools を既存エージェントの tools に設定し、Instructionには問い合わせ受付の手順を追加します

問い合わせの送信を依頼されたら、次の手順で対応してください。

1. 問い合わせ内容と連絡先メールアドレスを確認する。
2. 期待した挙動は任意で確認する。
3. 送信内容をまとめて提示し、保存してよいか確認する。
4. 利用者が明確に了承した場合だけsubmit_inquiryを呼び出す。
5. 内容が変更された場合は、変更後の内容で再度了承を得る。
6. savedの場合だけ受付完了と受付IDを伝える。
7. failed・unknown、または応答を受け取れない場合は、自動で再送しない。

Agent Id・Version・保存先は利用者に質問しない。

会話の中で不満や要望が出たという理由だけで送信せず、問い合わせとして送信する前に承認をもらうようにします

Workflow Builderから利用する場合

Workflow Builder側では、カスタムMCPデータストアのアクションを再読み込みし、submit_inquiry を有効にします

ヘッダーを指定できない接続では、エージェントの識別情報をURLクエリで渡します。

https://SERVICE.run.app/mcp?agent_id=helpdesk_agent&version=v1.0.0

ADKではヘッダー、Workflow BuilderではURLクエリパラメータという違いはありますが、MCPサーバー側では同じ処理を使用します

実際に問い合わせを送ってみる

実装できたら、テストで問い合わせを確認します

送るべき情報を聞いてくれました

必要な情報を入れてみます

確認をしてくれました

送信ボタン を押下します

ちゃんと保存されました

今後は、スクショを送れるようにしたいのと、
Workflow Builder 製エージェントの場合、 Agent Id と Version の指定方法が運用しづらい(MCPサーバーのURLクエリパラメータ)ので、改善していきたいと思います