はじめに #
MCPの勉強をしています。前回はDocker Desktop MCP Toolkitを使ってみました。
今回はMCPサーバーを実装してみたいと思います。プロンプトだけではうまく制御できない場合にMCPで制御できるといいことがありそうな気がします。
以下の情報を参照して作業を進めます。
MCP Python SDK #
概要 #
MCPによってアプリケーションからLLMに決められた方法でコンテキスト情報を渡すことができます。これによってコンテキスト情報とLLMとの連携を分離して扱うことができます。このPython SDKはMCPの仕様をすべて満たす実装がされていて、以下を簡単に実施できます。
- 任意のMCPサーバーに接続できるMCPクライアントを作る
リソース、プロンプト、ツール、を提供するMCPサーバーを作る- 標準出力や、ServerSentEventやStreamable HTTP等の標準的な通信を使います
- 全てのMCP プロトコルメッセージやライフサイクルイベントを制御します
インストール #
プロジェクトを作成します
uv init mcp-server-demo
cd mcp-server-demo
依存関係を追加します。
uv add "mcp[cli]"
以下でMCP開発ツールを起動する事ができます。
uv run mcp
やってみる #
計算ツールとデータを公開するシンプルなMCPサーバーを作ってみましょう。以下の内容のserver.pyを作成します。
# server.py
from mcp.server.fastmcp import FastMCP
# Create an MCP server
mcp = FastMCP("Demo")
# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"
参照先の記事では以下でClaude Desktopにインストールできるという説明がされています。
mcp install server.py
私は Cline 派なので、以下のようにして実行しました。まずmcp.run()を呼び出すようにコードを変更します。
+if __name__ == "__main__":
+ mcp.run()
そして、ClineのMCPサーバーに作成したpythonスクリプトを cline_mcp_settings.json に追加します。cline_mcp_settings.jsonについては前回記事を参照ください。このとき uv で作成したインタープリタをcommandに指定するところが注意点です。
{
... その他の設定
"demo-server": {
"autoApprove": [],
"disabled": false,
"timeout": 60,
"type": "stdio",
"command": "/Users/<user_name>/mcp-server-demo/.venv/bin/python",
"args": [
"/Users/<user_name>/mcp-server-demo/server.py"
]
}
}
}
以下のようにClineからMCPサーバーを呼び出すことができました。
あるいは以下のようにしてMCP Inspectorからも呼び出すこともできます。
mcp dev server.py
その他のMCPサーバーの例 #
Echoサーバー #
エコーするだけのサーバーです。
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Echo")
@mcp.resource("echo://{message}")
def echo_resource(message: str) -> str:
"""Echo a message as a resource"""
return f"Resource echo: {message}"
@mcp.tool()
def echo_tool(message: str) -> str:
"""Echo a message as a tool"""
return f"Tool echo: {message}"
@mcp.prompt()
def echo_prompt(message: str) -> str:
"""Create an echo prompt"""
return f"Please process this message: {message}"
SQLite Explorerサーバ #
SQLiteデータベースを探索するサーバーです。
データベースのテーブル情報を要求されたら get_schema が呼び出されて、テーブル情報を返します。
データベースのデータ操作についてLLMに依頼して返ってきたクエリを使ってquery_dataが呼び出されて実行します。
import sqlite3
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("SQLite Explorer")
@mcp.resource("schema://main")
def get_schema() -> str:
"""Provide the database schema as a resource"""
conn = sqlite3.connect("database.db")
schema = conn.execute("SELECT sql FROM sqlite_master WHERE type='table'").fetchall()
return "\n".join(sql[0] for sql in schema if sql[0])
@mcp.tool()
def query_data(sql: str) -> str:
"""Execute SQL queries safely"""
conn = sqlite3.connect("database.db")
try:
result = conn.execute(sql).fetchall()
return "\n".join(str(row) for row in result)
except Exception as e:
return f"Error: {str(e)}"
おわりに #
- MCPサーバーを実装してClineから呼び出すことができました。
- MCPサーバーのサンプルを確認しMCPサーバーの動作イメージを掴むことができました。 次はMCPサーバーの実装をもうちょっと独自で複雑なものにしてみたいと思います。
参考書籍 #
|  |
|  |
その他の情報 #
参照したgithubのREADMEを上から記載していこうと思ったのですが、どうもうまくまとめられないので、途中で全部記載するのは諦めました。参考までに以下の残骸の情報は残しておきます。
MCPサーバーって何？ #
Model Context Protocol (MCP)によってLLMアプリケーションに安全で統一された方法でデータと機能を公開することができます。 WebAPIのようなものですが、LLMとの連携のためだけに作られています。MCPサーバは以下を実行できます。
- Resources でデータを公開します。GETエンドポイントのようなもので、LLMのコンテキストに情報を渡すことができます
- Tools を使って機能を提供できます。POSTエンドポイントのようなもので、コードを実行したり副作用を持つことができます。
- Prompts を定義します。これによって再利用可能なテンプレートによるLLMとの連携が可能となります。
コアとなる概念 #
サーバ #
FastMCPサーバ はMCPプロトコルのコアとなるインタフェースです。接続管理、プロトコルの検証、メッセージルーティングを行います。
リソース #
Resource によってLLMへのデータの提供方法を定義します。REST API の GET エンドポイントのようなもので、副作用はありません。
ツール #
Tool によってLLMへの機能の提供方法を定義します。Resourceとことなり、コンピュート資源を使い副作用がありえます。
構造化された出力 ツールは戻り値の型アノテーションに互換性がある場合はデフォルトで構造化された結果を返し、そうではない場合は非構造化データを返します。
構造化された出力として以下をサポートしています。
- Pydantic モデル(BaseModelのサブクラス)
- TypedDict
- Dataclass及び型ヒント付きのクラス
- dict[str, T] (TはJSONシリアライズ可能な型)
- プリミティブ型及びジェネリック型。
{"result":value}で囲まれて返ります
型ヒントがないクラスはシリアライズできません。適切なアノテーションがあるクラスだけPydanticモデルに変換されスキーマ生成と検証に使われます。
構造化された結果は型ヒントから生成されたスキーマで自動検証されます。これによりクライアントが簡単に読めるデータを返すことが保証できます。
:::message
非構造化データが返ることがあるのは、古いバージョンのMCP仕様でも使えるようにする(後方互換性を保つ)ためです。
ツール関数の戻り値の型アノテーションが構造化データとして分類されるがそれが望ましくない場合は@toolデコレータでstructured_output=Falseを指定してください。
:::
プロンプト #
LLMとMCPサーバのやりとりを助ける再利用可能なテンプレートです。
画像データ #
FastMCPは画像データを操作できる Image クラスを持ちます。PillowのImageオブジェクトを返却できるということですね。
コンテキスト #
コンテキストはツールとリソースにMCPサーバにアクセスする機能を付与します。
補完機能 #
MCPはプロンプトの引数の補完候補の提供をサポートします。コンテキスト情報を使い、サーバーは以前使用した値を元に保管情報を提供します。
誘導 #
ツールの実行中にユーザに対して追加の情報を要求します。elicit() 関数は ElicitationResult を返します。ElicitationResult は 以下の要素を含みます。
action: “accept”、“decline”、あるいは “cancel” のいずれかdata: 検証済みのレスポンス。actionが “accept” の場合のみvalidation_error: 検証エラー
認証 #
認証は、保護されたリソースにアクセスするツールを公開したいサーバーで使用できます。
mcp.server.auth はOAuth2.1のリソースサーバーの機能を実装します。MCPサーバーは分離された認証サーバーで発行されたトークンを検証するリソースサーバーとして動作します。これはMCP authorization specificationに従ったもので、認証サーバディスカバリで使うRFC9728(Protected Resource Metadata)を実装します。
MCPサーバは TokenVerifier プロトコル実装することで認証機能を使うことができます。
アーキテクチャ #
- 認証サーバ: OAuthフローを使い、ユーザ認証をしてトークンを払い出します
- リソースサーバー: トークンを検証し保護されたリロースを提供します
- クライアント: RFC9728を通して認証サーバを見つけ、トークンを取得しMCPサーバで使用します
トークンの検証についてはTokenVerifierを参照ください。
サーバーを実行する #
開発モード #
MCP Inspector を使うのが最も早い方法です。
mcp dev server.py
# Add dependencies
mcp dev server.py --with pandas --with numpy
# Mount local code
mcp dev server.py --with-editable .
Claude Desktop を使う場合 #
以下でインストールできるそうです。私はCline派なので試していません。
mcp install server.py
# Custom name
mcp install server.py --name "My Analytics Server"
# Environment variables
mcp install server.py -v API_KEY=abc123 -v DB_URL=postgres://...
mcp install server.py -f .env
直接実行 #
カスタムデプロイなどの高度なシナリオの場合は以下。Clineで使う場合はこちらの使い方になります。以下のように実装します。
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
if __name__ == "__main__":
mcp.run()
以下で実行します。
python server.py
# or
mcp run server.py
:::message
mcp run や mcp dev は FastMCP を使った場合のみ使えます。
:::
Streamable HTTP 転送 #
:::message 本番環境でのデプロイにおいては、ストリーム対応のHTTPトランスポートがSSEトランスポートに代わりつつあります。 ::: FastMCPサーバはFastAPIアプリケーション上で複数マウントすることができます。 Streamable HTTPを使った低レベル実装は以下に情報があります。
Streamable HTTP転送は以下をサポートしています。
- ステートフルモードとステートレスモード
- イベントストアのやり直し
- JSONあるいはSSEレスポンスフォーマット
- マルチノードデプロイ向けのスケーラビリティ
既存のASGIサーバーにマウントする #
:::message こちらも、ストリーム対応のHTTPトランスポートがSSEトランスポートに代わりつつあります。 :::
デフォルトでは SSEサーバは /sse にマウントされ、Streamable HTTPは /mcp にマウントされます。パスは以降で紹介する方法で変更できます。
まず、sse_app関数で既存のASGIサーバーにSSEサーバーをマウントできます。これによりSSEサーバーをその他のASGIアプリに統合できます。
また、複数のMCPサーバーを異なるパスにマウントすることもできます。Starletteを使う例についてはこちらを参照ください。