python+geminiだけで動く単純なMCPサーバを作ってみる
MCPサーバとクライアントを作ってみる
世の中にあるMCPサーバの作り方の記事を読んでも私の理解力が低すぎるからだと思うが、Claude Desktopとの連携だったりGoogle Colaboを使っていたりとかであんまり良く理解できなかったので、物凄くシンプルなものから作ってみる。
作りたいのはMCPのクライアントからMCPのサーバに問い合わせを投げて最終的にMCPのクライアントから結果を返してもらうっていう単純なもの。
Claude Desktopとはは使わず、純粋にpythonだけで完結するようにする。
作ってみたらなんとなくがだ理解が捗った気がする。
でも、よくわからないのがLLMが何度もMCPサーバを呼び出して最終的に結果を取得するみたいなことをしたいとした場合、どうやってLLMを何度も呼び出せばいいのか、また結論が出たと判断するタイミングをどうするのか?がまだよくわかっていないので、もうしばらく触ってみたいと思う。
流行っているだけあって、便利そうだなとは思った。
最終版のシーケンス図
sequenceDiagram participant User participant Gemini participant ClientApp participant MCPServer participant greetFunction participant farewellFunction User->>ClientApp: 名前と時刻を入力(例: "Alice", "10:00") ClientApp->>Gemini: 挨拶してほしい内容と使える機能リストをAIに送る Note right of Gemini: 例:「今の時刻は10:00です。Aliceさんに挨拶してください」\n(使える機能: greet, farewell) Gemini-->>ClientApp: 選択した機能(function_call結果: greet, {"name": "Alice"})を返す ClientApp->>MCPServer: greet, {"name": "Alice"} MCPServer->>greetFunction: greet("Alice") greetFunction-->>MCPServer: "Hello, Alice!" MCPServer-->>ClientApp: "Hello, Alice!" ClientApp-->>User: "Hello, Alice!" alt 午後の場合 User->>ClientApp: 名前と時刻を入力(例: "Alice", "15:00") ClientApp->>Gemini: 挨拶してほしい内容と使える機能リストをAIに送る Note right of Gemini: 例:「今の時刻は15:00です。Aliceさんに挨拶してください」\n(使える機能: greet, farewell) Gemini-->>ClientApp: 選択した機能(function_call結果: farewell, {"name": "Alice"})を返す ClientApp->>MCPServer: farewell, {"name": "Alice"} MCPServer->>farewellFunction: farewell("Alice") farewellFunction-->>MCPServer: "Goodbye, Alice!" MCPServer-->>ClientApp: "Goodbye, Alice!" ClientApp-->>User: "Goodbye, Alice!" end
仕様
- python3.13
- fastmcp 2.5.2
- google-genai 1.18.0
- uv
【接続の確認】いちばん簡単なMCPサーバとクライアント
実行方法
起動は以下のコマンドで実行する。
mcp_simple_server.pyは実行しなくても良い。
今回の方法の場合はmcp_simple_client.pyを起動するとmcp_simple_server.pyをサブプロセスでクライアンが起動してくれる。
geminiを使うサンプルはgeminiのAPIキーが必要なので取得して、環境変数にGEMINI_API_KEYの名称で登録してください。
$ uv init $ uv add google-genai $ uv add fastmcp $ uv run mcp_simple_client.py Client connected: True
MCPサーバ
- mcp_simple_server1.py
from fastmcp import FastMCP mcp: FastMCP = FastMCP("My MCP Server") if __name__ == "__main__": mcp.run()
MCPクライアント
- mcp_simple_client1.py
import asyncio from pathlib import Path from fastmcp import Client version = Path(__file__).name.split(".py")[0][-1] server = "mcp_simple_server{}.py".format(version) client: Client = Client(server) async def main(): async with client: print(f"Client connected: {client.is_connected()}") if __name__ == "__main__": asyncio.run(main())
【基本】MCPクライアントからMCPサーバへリクエストをする
MCPクライアントからMCPサーバへのリクエストは基本的にmcp.toolに対して行われる。
これだとMCPクライアントってMCPサーバのtoolを呼び出すだけだから何が便利なんだけってなる。
便利になるのはMCPクライアントのmainにgeminiを入れてgeminiにMCPサーバのどのtoolを呼び出すのか決めさせることができるところかな?
ここでは単純にMCPクライアントからMCPサーバのtoolを自分で呼んでる。
以下のようなレスポンスが取得できる。
Client connected: True Tool result: [TextContent(type='text', text='Hello, Alice!', annotations=None)]
サーバ
from fastmcp import FastMCP from rich import print mcp: FastMCP = FastMCP("My MCP Server") @mcp.tool() def greet(name: str) -> str: return f"Hello, {name}!" if __name__ == "__main__": mcp.run()
クライアント
import asyncio from pathlib import Path from fastmcp import Client version = Path(__file__).name.split(".py")[0][-1] server = "mcp_simple_server{}.py".format(version) client: Client = Client(server) async def main(): async with client: print(f"Client connected: {client.is_connected()}") # MCP サーバで定義した@map.tool()のdefの名称を指定 tool_name = "greet" # toolに渡すパラメータを指定 tool_args = {"name": "Alice"} tool_result = await client.call_tool(tool_name, tool_args) print(f"Tool result: {tool_result}") if __name__ == "__main__": asyncio.run(main())
【応用】MCPクライアントにgeminiを入れてgeminiにMCPサーバのtoolの使い分けをさせる
単純なもののほうがシンプルでいいよねと思うので以下の仕様とする。
MCPクライアントにgeminiを入れて、geminiに時間帯によって挨拶とさようならを使い分けさせる。
余談だけどgeminiのAPIキーは無料で取得できるのでちょっと使うなら便利なので取っておくと良いと思う。
取り方は最後に記載する。
サーバ
こっちは説明するほどの内容もないが、MCPクライアントから呼び出せる機能が2個ある。
greetが呼び出されたらHelloを返し、farewellが呼び出されたらGoodbyを返すという単純なもの。
from fastmcp import FastMCP mcp: FastMCP = FastMCP("My MCP Server") @mcp.tool() def greet(name: str) -> str: return f"Hello, {name}!" @mcp.tool() def farewell(name: str) -> str: return f"Goodbye, {name}!" if __name__ == "__main__": mcp.run()
クライアント
一気にコードの量が増えてややこしくなるが、なるべく機能はシンプルにして理解しやすいように書いてみた。
仕様
MCPクライアントが実行されるときに名前を渡すと実行された時間帯によって適切な挨拶が返ってくるという機能。 プログラムではMCPサーバのどの機能を呼び出すのかは決定せず、Geminiに渡すプロンプトに含まれる時間を判断して、greet, farewellのどちらを呼び出すのかを決める。 その後、MCPクライアントがMCPサーバを呼び出してレスポンスを取得して、MCPクライアントとして挨拶文を返す。 詳細な動作は上の方に書いたシーケンス図を見たほうが分かりやすいと思う。
import asyncio import os from datetime import datetime from pathlib import Path import mcp from fastmcp import Client from google import genai from google.genai import types GEMINI_API_KEY: str = os.environ.get("GEMINI_API_KEY") # type:ignore GEMINI_MODEL = "gemini-2.0-flash" if not GEMINI_API_KEY: raise ValueError("GEMINI_API_KEY is not set") version = Path(__file__).name.split(".py")[0][-1] server = "mcp_simple_server{}.py".format(version) client: Client = Client(server) class Gemini: """ Gemini LLMとの連携およびfunction callingインターフェースを提供するクラス。 """ def __init__(self, api_key: str = GEMINI_API_KEY, model: str = GEMINI_MODEL) -> None: """ Geminiクライアントを初期化する。 Args: api_key (str): Gemini APIキー。 model (str): 利用するGeminiモデル名。 """ # ここはGeminiがMCPサーバのどのtoolを呼び出すのか決めるための定義とtoolを呼び出すときの定義 # LLMによって定義方法が異なるので利用するLLMに合わせて修正が必要 function_declarations = [ { "name": "greet", "description": "午前中(0時から11時59分まで)に使う、相手への挨拶を返すツールです。", "parameters": { "type": "object", "properties": { "name": { "type": "string", "description": "相手の名前", }, }, "required": ["name"], }, }, { "name": "farewell", "description": "午後(12時から23時59分まで)に使う、相手への別れの挨拶を返すツールです。", "parameters": { "type": "object", "properties": { "name": { "type": "string", "description": "相手の名前", }, }, "required": ["name"], }, }, ] self.tool_defs = types.Tool(function_declarations=function_declarations) # type: ignore self.gemini = genai.Client(api_key=api_key) self.model = model def choose_function(self, prompt: str) -> tuple: """ Geminiモデルにプロンプトを渡し、function callingで呼ぶべきツール名と引数を抽出する。 Args: prompt (str): LLMへの入力プロンプト。 Returns: tuple: (tool_name, tool_args) のタプル。 Raises: ValueError: function_callが見つからない場合。 """ response = self.gemini.models.generate_content( model=self.model, contents=[prompt], config=types.GenerateContentConfig(temperature=0, tools=[self.tool_defs]), ) tool_name, tool_args = self.parse_function_call(response) return tool_name, tool_args def parse_function_call(self, response: types.GenerateContentResponse): """ Geminiから返されたレスポンスからfunction_callを抽出する。 Args: response (types.GenerateContentResponse): Geminiのレスポンス。 Returns: tuple: (tool_name, tool_args) Raises: ValueError: function_callが見つからない場合。 """ # Geminiが決めたfunction callingで呼び出すべきtool_nameとパラメータを返す for candidate in response.candidates: # type: ignore for part in candidate.content.parts: # type: ignore if hasattr(part, "function_call") and part.function_call: func_call = part.function_call tool_name = func_call.name # type: ignore tool_args = func_call.args # type: ignore return tool_name, tool_args raise ValueError("No function_call found in Gemini response") class MCP: """ MCPサーバへのツール呼び出しをラップするクラス。 """ def __init__(self, client: Client) -> None: """ MCPクライアントの初期化。 Args: client (Client): fastmcpのClientインスタンス。 """ self.client = client async def call_tool(self, tool_name: str, tool_args: dict) -> str | None: """ MCPサーバにツール呼び出しを行い、テキストレスポンスを返す。 Args: tool_name (str): 呼び出すツール名。 tool_args (dict): ツールへ渡す引数。 Returns: str | None: ツールのテキストレスポンス(TextContentのみ対応)。 """ async with self.client: # MCPサーバのツールを呼び出す response = await self.client.call_tool(tool_name, tool_args) res = self.parse_response(response) return res def parse_response(self, response: list) -> str | None: """ MCPサーバから返されたレスポンスリストからテキストのみ抽出。 Args: response (list): MCPサーバのレスポンス。 Returns: str | None: 最初のTextContent.text、なければNone。 """ for res in response: if type(res) is mcp.types.TextContent: return res.text return None class Coordinator: """ GeminiおよびMCPクラスを連携させ、指定した名前に応じた挨拶文などを生成するコーディネータ。 """ def __init__(self, gemini: Gemini, mcp: MCP) -> None: """ Coordinatorの初期化。 Args: gemini (Gemini): Geminiクライアント。 mcp (MCP): MCPクライアント。 """ self.gemini = gemini self.mcp = mcp async def get_message(self, name: str, now: str | None = None) -> str | None: """ 指定した名前・時刻に基づき、Gemini→MCPサーバを連携して挨拶文などを取得する。 Args: name (str): 相手の名前。 now (str | None): 現在時刻(HH:MM形式、省略時は現在時刻を自動設定)。 Returns: str | None: MCPツールによる挨拶などのメッセージ。 """ if now is None: now = datetime.now().strftime("%H:%M") prompt = f"現在時刻は{now}です。{name}さんに挨拶をしてください。" tool_name, tool_args = self.gemini.choose_function(prompt) mcp_res = await self.mcp.call_tool(tool_name, tool_args) return mcp_res async def main(): """ サンプル全体の実行エントリポイント。 """ gemini = Gemini() mcp = MCP(client) coordinator = Coordinator(gemini, mcp) res = await coordinator.get_message("Alice", "15:00") print(res) if __name__ == "__main__": asyncio.run(main())
GeminiのAPIキー
Geminiは少しぐらいなら無料でAPIが使えるのでちょっとしたテストに便利だと思う。
Google AI Studio に行くと上に方にGet API KEYというのがあるので、