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
from fastmcp import FastMCP
mcp: FastMCP = FastMCP("My MCP Server")
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()}")
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()}")
tool_name = "greet"
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に時間帯によって挨拶とさようならを使い分けさせる。
余談だけど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")
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モデル名。
"""
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)
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が見つからない場合。
"""
for candidate in response.candidates:
for part in candidate.content.parts:
if hasattr(part, "function_call") and part.function_call:
func_call = part.function_call
tool_name = func_call.name
tool_args = func_call.args
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:
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というのがあるので、
APIキーを作成を押すとAPIキーが発行できる。
