記事概要

筆者がMCPについて調べて理解した内容を備忘としてメモした内容である。
記事の後半には、MCPのサーバとクライアントをpythonとDockerで動かせるように作ったサンプルコードを掲載している。

MCPの概要

MCP（Model Context Protocol）は、HTTPのようなプロトコルである。
- HTTP：決められたルールがあるので、それを
- MCP：LLMが外部ツールやデータソースと安全に連携するためのプロトコル。HTTPの場合は、クライアント → サーバへの一方通行のやりとりで合ったが、MCPの場合は、クライアント ↔︎ サーバの双方向でのやり取りができる。
  - 双方向でのやり取りができるのは、外部ツールがLLMを利用したいシチュエーションがあるため。
    - 例）
      - クライアントが、特定のデータ分析を行いたい、と要求した。
        
        この時、サーバがすべきことは、テーブルやカラム、データの中身を洗い出して、その分析に必要なものを選定すること。
        
        サーバは、メタデータがあるので、これらの情報を洗い出すことはできても、それが分析ニーズを満たすために使えるものかどうかのジャッジはできない。
        
        そこで、LLMにジャッジしてもらう必要が出てくる。
        
        このジャッジしてもらうためには、サーバ → クライアントへの要求を出す必要がある。だから、双方向的な通信ができる点が一つの特徴と言える。
MCPと関連する技術用語を先にイメージすると理解しやすい。
- 技術用語
  - HTTP
  - API（特にREST API）
そして、MCPとはなんぞや？を理解するためのミソは、
- MCPはAPIを「ラップ」し、LLMが理解できる形式で機能を提供する
- MCPの多くの実装では、APIとの通信はHTTPベースで行われているが、それをLLMにとって使いやすいインターフェースに変換しているのがMCP
- APIは一方通行が前提ではあるが、MCPの場合は上述の通り、サーバとクライアントが双方向でやり取りできる（というかLLMが自律的にタスクを実行するためにはそうするしかない）点が特徴的。
実際にMCPサーバを”作る”側は、FastAPIっぽく構築することができる（関数にデコレータつけてAPIを作る感じが似てる）。クライアント側はサーバクライアントシステムの”クライアント”を作る感じ。
しかし、内部的には双方向的なやり取りやLLMが動的にタスク状況を理解したり文脈を理解するための各種制御が含まれている。

作り方

MCPの考え方として、クライアント-サーバ形式に近い形態となっている。
そのため、MCPを作るとなったら、クライアント側を作るのか、サーバ側を作るのかで変わってくる
今回は、ひとまずサーバ側とサーバを簡易的に呼び出すだけのクライアントを作成してみる。
- クライアント側のコードを作成するのは結構複雑。
- 参考）https://modelcontextprotocol.io/quickstart/client

環境構築

ファイル構成

.
├src/
　┗ server.py
　┗ client.py
├── Dockerfile
├── docker-compose.yaml
├── Makefile
└── .env

Dockerfile

FROM python:3.11-slim

# Build-time arguments
ARG POETRY_VERSION
ARG POETRY_HOME
ARG USER_UID
ARG USERNAME

# 必要なパッケージをインストール（curl等）
RUN apt-get update && apt-get install -y --no-install-recommends \
    curl \
    && apt-get clean \
    && rm -rf /var/lib/apt/lists/*

# Poetryのインストール
RUN echo ${POETRY_HOME}
SHELL ["/bin/bash", "-o", "pipefail", "-c"]
RUN curl -sSL https://install.python-poetry.org/ | python3 - --version ${POETRY_VERSION} && \
    ln -s ${POETRY_HOME}/bin/poetry /usr/local/bin/poetry

# Node.jsとnpmのインストール
RUN curl -fsSL https://deb.nodesource.com/setup_18.x | bash - && \
apt-get install -y --no-install-recommends nodejs && \
apt-get clean && \
rm -rf /var/lib/apt/lists/*

# ==========================================
# Create user in the container to avoid permission matter 
# incompatible between host and container user
# ==========================================
RUN useradd --uid ${USER_UID} -m ${USERNAME} 

USER $USERNAME
ENV PATH="/usr/local/bin:$PATH"

WORKDIR /home/${USERNAME}/

docker-compose.yaml

services:
  python-dev: # サービス名 (任意)
    build:
      context: . # Dockerfileがあるディレクトリ (カレントディレクトリ)
      dockerfile: Dockerfile
      args:
        POETRY_VERSION: ${POETRY_VERSION}
        POETRY_HOME: ${POETRY_HOME}
        USER_UID: ${USER_UID}
        USERNAME: ${USERNAME}
    container_name: ${CONTAINER_NAME}
    volumes:
      - type: bind
        source: ./src
        target: /home/${USERNAME}
    working_dir: /home/${USERNAME}
    ports:
      - "6274:6274"  # MCPサーバのポートをホストに公開
      - "6277:6277"  # MCPサーバのポートをホストに公開
    tty: true 
    restart: always # コンテナが停止した場合に常に再起動する

.env

CONTAINER_NAME=mcp-learning
POETRY_VERSION=2.1.2
POETRY_HOME=/opt/poetry

Makefile

include .env
export

.PHONY: build \
	up \
	down \
	shell \
	write_uid_onto_env

# Dockerイメージのビルド
build:
	@ docker compose build

# コンテナを起動
up:
	@ docker compose up -d

# コンテナの停止と削除
down:
	@ docker compose down

# ホストマシンのUIDを.envに書き込む
# 一時ファイルに避難させておかないと、USER_UIDやUSERNAMEがずっと追記され続けてしまうので、
# このような実装とした。
write_uid_onto_env:
	@ TEMP_FILE=$$(mktemp); \
	grep -v "^USER_UID=" .env | grep -v "^USERNAME=" > $$TEMP_FILE; \
	echo "USER_UID=$(shell id -u $(shell whoami))" >> $$TEMP_FILE; \
	echo "USERNAME=$$USER" >> $$TEMP_FILE; \
	mv $$TEMP_FILE .env

shell:
	@ $(MAKE) write_uid_onto_env
	@ $(MAKE) build
	@ $(MAKE) up
	@ docker exec -it "$(CONTAINER_NAME)" /bin/bash

（サーバ側の）コード

server.py

# 最小限のサーバをたてる

from mcp.server.fastmcp import FastMCP

# サーバインスタンスの作成
mcp = FastMCP("minimal-mcp-server")

# ---------------------------------------------------------------------
# MCPサーバの機能には大きく3種類ある
# それぞれの機能は、サーバに登録することで利用できるようになる
# - Prompts: サーバがクライアントのLLMを使いたい時にプロンプトを送るためのもの
# - Resources: クライアントがサーバのデータリソースを使いたい時にリソースを取得するためのもの。例：リポジトリの情報を取得する
# - Tools: クライアントがサーバのツールを使いたい時にツールを実行するためのもの。例：SQLクエリを実行する

# デコレータを使うことで、関数をツールとして登録できる
# 関数のドキュメント文字列は、ツールの説明として使用される.
# 関数のパラメータと戻り値の型アノテーションは、自動的にMCPの型情報に変換される.
# ---------------------------------------------------------------------

#################################
# ツールの定義
# エコーツールの定義
@mcp.tool()
def echo(message: str) -> str:
    """
    入力されたメッセージ（str）をそのまま返すツール
    """
    return f"Echo: {message}"

# 現在日時を返すツール
@mcp.tool()
def current_time() -> str:
    """
    現在の日時を返すツール
    """
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

#################################
# リソースの定義
@mcp.resource("info://server")
def server_info() -> str:
    """
    サーバの情報を返すリソース
    """
    return "This is a minimal MCP server."

@mcp.resource("greeting://{name}")
def greeting(name: str) -> str:
    """
    名前を受け取って挨拶を返すリソース
    """
    return f"Hello, {name}!"

#################################
# プロンプトの定義
@mcp.prompt()
def simple_prompt(text: str) -> str:
    """
    シンプルなプロンプト
    """
    return f"以下のテキストについて考えて答えてください: {text}"

# 直接実行するときのメイン実行部分
if __name__ == "__main__":
    mcp.run(host="0.0.0.0", port=6274)

テスト動作させる

MCP Inspectorを使ってサーバをテストする

poetry run mcp dev server.py

実行すると、以下のようなブラウザが立ち上がる。

上の方にある
- Resources / Prompts / Tools　が自作で定義した各種ツールである。
- このようにして、自作のツールの動作をブラウザで直接確かめることができる

（クライアント側の）コード

サーバ側のコードは、FastAPIのように、デコレータをつけてAPIっぽい構築方法ができた。
一方で、クライアント側は、自分でクライアントを作る必要があり、そこが複雑。今回はChatGPTに作ってもらった。
- 実際にMCPを使うときは、VSCodeやClaudeDesktopのようなツールを経由して利用することが多いと思われる。そういったツールはツール自体にクライアント機能を有しているので、このような実装を自前でする必要はない。

client.py

import asyncio
from contextlib import AsyncExitStack

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class SimpleMCPClient:
    def __init__(self):
        self.session = None
        self.exit_stack = AsyncExitStack()

    async def connect(self, server_script_path: str):
        server_params = StdioServerParameters(
            command="python",
            args=[server_script_path],
            env=None
        )
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        self.stdio, self.write = stdio_transport
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
        await self.session.initialize()
        print("✅ Connected to MCP Server!")

    async def call_tool(self, tool_name: str, arguments: dict):
        result = await self.session.call_tool(tool_name, arguments)
        return result.content

    async def get_resource(self, resource_uri: str):
        result = await self.session.read_resource(resource_uri)
        return result.contents

    async def run(self):
        # ツール一覧確認
        response = await self.session.list_tools()
        print("📦 Available tools:", [tool.name for tool in response.tools])

        # ----- テスト：ツール呼び出し -----
        print("----- テスト：ツール呼び出し -----")
        echo_result = await self.call_tool("echo", {"message": "Hello MCP!"})
        print(f"🗨️ Echo Tool Result: {echo_result}")

        time_result = await self.call_tool("current_time", {})
        print(f"⏰ Current Time Tool Result: {time_result}")

        for _ in range(3):
            print()

        # ----- テスト：リソース取得 -----
        print("----- テスト：リソース取得 -----")
        server_info = await self.get_resource("info://server")
        print(f"ℹ️ Server Info Resource: {server_info}")

        greeting = await self.get_resource("greeting://Alice")
        print(f"👋 Greeting Resource: {greeting}")

        for _ in range(3):
            print()

        # ----- テスト：プロンプト実行 -----
        print("----- テスト：プロンプト実行 -----")
        prompt_result = await self.session.get_prompt(
            "simple_prompt",
            {"text": "明日の天気を予想してください。"}
        )
        print(f"🧠 Prompt Result: {prompt_result}")

        for _ in range(3):
            print()

    async def cleanup(self):
        await self.exit_stack.aclose()

async def main():
    server_script = "server.py"  # サーバースクリプトのパス
    client = SimpleMCPClient()

    try:
        await client.connect(server_script)
        await client.run()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

作成したコードを動かしてみる。

ターミナルが2つ必要である。

ターミナル①：サーバを起動させておく
ターミナル②：クライアントを動かす。

ターミナル①

以下のコマンドを打てばOK

poetry run python server.py

ターミナル②

以下のコマンドを打てばOK

poetry run python client.py