【新人が Google Cloud を触ってみた】MCP + Cloud Run で 文書検索エージェント を作ってみた

{
  "documents": [
    {
      "id": "DOC-003",
      "title": "社内システムメンテナンスのお知らせ（8月）",
      "category": "システムメンテナンス",
      "published_date": "2026-08-10",
      "tags": ["メンテナンス", "システム", "停止"],
      "summary": "2026年8月20日 22:00から翌1:00まで、社内ポータルサイトを停止してメンテナンスを実施します。"
    }
  ]
}  

def search_documents(keyword: str) -> list[dict]:
    """タイトル・本文抜粋・タグのいずれかにキーワードを含む文書を検索する。"""
    keyword_lower = keyword.lower()
    results = []
    for doc in _DOCUMENTS:
        haystack = " ".join(
            [doc["title"], doc["summary"], " ".join(doc["tags"])]
        ).lower()
        if keyword_lower in haystack:
            results.append(doc)
    return results  

import os
from mcp.server.fastmcp import FastMCP
import search_logic

PORT = int(os.environ.get("PORT", 8080))  # Cloud Runが指定するポート番号
mcp = FastMCP(name="document-search-server", host="0.0.0.0", port=PORT)


@mcp.tool()
def search_documents(keyword: str) -> dict:
    """タイトル・本文・タグにキーワードを含む発信文書を検索する。"""
    results = search_logic.search_documents(keyword)
    return {"count": len(results), "documents": results} 

if __name__ == "__main__":
    mcp.run(transport="streamable-http")  # Cloud Run上ではHTTP経由で受けるため
  

import asyncio
import streamlit as st
from google import genai
from google.genai import types
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

# --- ① MCPのツール定義をGeminiのFunctionDeclaration形式に変換 ---
def mcp_tool_to_gemini_function(tool):
    return types.FunctionDeclaration(
        name=tool.name, description=tool.description or "", parameters=tool.inputSchema
    )

# --- ② 質問を受けて、Gemini呼び出しとMCPツール実行をループ処理 ---
async def ask_once(client, session, gemini_tool, contents, user_prompt):
    contents.append(types.Content(role="user", parts=[types.Part.from_text(text=user_prompt)]))

    for turn in range(5):
        # Geminiに問い合わせ（MCPから取得したツール定義を渡す）
        response = client.models.generate_content(
            model=MODEL_NAME, contents=contents,
            config=types.GenerateContentConfig(tools=[gemini_tool]),
        )
        model_content = response.candidates[0].content
        contents.append(model_content)

        # ツール呼び出し指示（function_call）がなければ最終回答として終了
        function_call_parts = [p for p in model_content.parts if p.function_call]
        if not function_call_parts:
            st.chat_message("assistant").write(response.text)
            return

        # ツール呼び出し指示がある場合、MCP Serverを実行
        for part in function_call_parts:
            fn_call = part.function_call
            # ★ MCP Clientを通じてリモートのMCP Serverのツールを呼び出す
            tool_result = await session.call_tool(fn_call.name, dict(fn_call.args))
            result_text = tool_result.content[0].text if tool_result.content else "該当なし"
            
            # 実行結果をGeminiに返すためのPartを作成
            contents.append(types.Content(role="user", parts=[
                types.Part.from_function_response(name=fn_call.name, response={"result": result_text})
            ]))

# --- ③ MCP Serverへ接続し、ツール一覧を取得して会話を開始 ---
async def process_chat(user_input):
    client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)
    
    # Cloud Run上のMCP ServerへHTTP接続
    async with streamablehttp_client(MCP_SERVER_URL) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # ★ MCP Serverから利用可能なツール一覧を取得
            tools_result = await session.list_tools()
            gemini_tool = types.Tool(
                function_declarations=[mcp_tool_to_gemini_function(t) for t in tools_result.tools]
            )
            
            # Agentとのやり取りを実行
            await ask_once(client, session, gemini_tool, st.session_state.contents, user_input)

# （※以下、StreamlitのUI描画処理およびセッション管理処理は省略）

実行確認