skydum

個人的な作業記録とか備忘録代わりのメモ

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

仕様

  1. python3.13
  2. fastmcp 2.5.2
  3. google-genai 1.18.0
  4. 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というのがあるので、

APIキーを作成を押すとAPIキーが発行できる。