Python MCPサーバ開発入門 - LLMアプリケーションとコンテキストを接続する

Python MCPサーバ開発入門

はじめに

最近のAI開発において、大規模言語モデル（LLM）の活用が急速に広がっています。しかし、LLMを既存のデータや機能と連携させるには、個別にインテグレーションを開発する必要があり、時間とリソースがかかります。

そこで登場したのが**Model Context Protocol (MCP)**です。MCPは、LLMと外部データソースやツールとの連携を標準化するオープンプロトコルです。本記事では、PythonでMCPサーバを開発する方法を解説し、LLMアプリケーションのための強力なインテグレーションを簡単に構築できるようになることを目指します。

目次

1. Model Context Protocolとは
2. MCPのアーキテクチャと概念
3. 開発環境の準備
4. 基本的なMCPサーバの作成
5. リソース、ツール、プロンプトの実装
6. 実践的なMCPサーバの例
7. クライアントとの連携
8. デプロイと運用
9. まとめと次のステップ

1. Model Context Protocol（MCP）とは

Model Context Protocol（MCP）は、AI・LLMアプリケーションと外部データソースやツールとの間をシームレスに接続するためのオープンプロトコルです。LLMは優れた推論能力を持っていますが、外部の最新データや特定のアプリケーション機能にアクセスするためには、専用のインテグレーションが必要でした。

MCPは「AIアプリケーションのためのUSB-C」とも呼ばれ、標準化されたインターフェースを提供することで、さまざまなデータソースやツールとLLMを簡単に接続できるようにします。

MCPの主なメリット

• 標準化されたインターフェース: 一度作成したMCPサーバは、あらゆるMCP対応クライアントと連携可能
• 分離された関心事: コンテキスト提供とLLMとのやり取りを分離し、モジュール性を向上
• セキュリティ: ローカルデータへのアクセスを安全に制御可能
• 再利用性: 既存のデータソースやAPIに対するMCPラッパーを作成して共有可能

2. MCPのアーキテクチャと概念

MCPは以下のコンポーネントから構成されています：

出典: https://modelcontextprotocol.io/

主要コンポーネント

• MCPホスト: Claude Desktopなど、LLMと連携しMCPサーバにアクセスするアプリケーション
• MCPクライアント: サーバとの1対1の接続を維持するプロトコルクライアント
• MCPサーバ: 特定の機能を標準化されたプロトコルで公開する軽量プログラム
• データソース: サーバがアクセスするローカルまたはリモートのデータソース

基本コンセプト

MCPプロトコルは3つの基本的な「プリミティブ」を定義しています：

1. リソース（Resources）: データやコンテンツをLLMに提供するための仕組み（GET的な概念）
2. ツール（Tools）: LLMに機能を提供する仕組み（POST的な概念）
3. プロンプト（Prompts）: LLMとの対話パターンを定義する再利用可能なテンプレート

3. 開発環境の準備

まず、MCPサーバを開発するための環境を整えましょう。

必要なツールのインストール

Python 3.8以上と、公式のMCP SDKが必要です。

# pipを使用する場合
pip install mcp

# uvを使用する場合（推奨）
uv add "mcp[cli]"

プロジェクト構成

MCPサーバのプロジェクト構成は比較的シンプルです。単一のPythonファイルでも、複雑なプロジェクト構造でも動作します。

以下は基本的なプロジェクト構成の例です：

my_mcp_server/
├── server.py      # メインのMCPサーバコード
├── pyproject.toml # プロジェクト設定（オプション）
└── README.md      # ドキュメント（オプション）

4. 基本的なMCPサーバの作成

最もシンプルなMCPサーバを作成してみましょう。以下は「エコー」機能を提供する基本的なサーバの例です。

"""
Simple Echo MCP Server
This example demonstrates a basic MCP server that exposes a simple echo tool.
"""
from mcp.server.fastmcp import FastMCP

# MCPサーバを作成
mcp = FastMCP("Echo Server")

# テキストをエコーバックするツールを定義
@mcp.tool()
def echo(text: str) -> str:
    """Echo the input text back to the caller."""
    return f"Echo: {text}"

# サーバを実行（スクリプト直接実行時）
if __name__ == "__main__":
    # デフォルトでstdioトランスポートを使用してサーバを起動
    mcp.run()

このシンプルなサーバは:

1. FastMCPクラスを使ってMCPサーバを初期化
2. @mcp.tool()デコレータを使ってツールを定義
3. mcp.run()でサーバを起動

です。

サーバの実行方法

作成したスクリプトは以下のように実行できます：

# 直接実行
python simple_echo_server.py

# MCPコマンドライン経由
mcp run simple_echo_server.py

# 開発モード（MCP Inspectorで動作確認）
mcp dev simple_echo_server.py

5. リソース、ツール、プロンプトの実装

MCPの3つの基本要素について詳しく見ていきましょう。

リソース（Resources）の実装

リソースはLLMに情報を提供するための仕組みです。RESTfulAPIのGETエンドポイントに似ています。

# 静的なリソースの定義
@mcp.resource("echo://static")
def static_echo() -> str:
    """Provide a static echo message."""
    return "This is a static echo message from the MCP server!"

# 動的パラメータを持つリソースの定義
@mcp.resource("echo://{message}")
def dynamic_echo(message: str) -> str:
    """Echo the message specified in the resource URI."""
    return f"Resource Echo: {message}"

リソースは:

• URI形式のパスで定義（scheme://path）
• 動的パラメータをパス内に{}で指定可能
• 返り値はテキスト、画像、埋め込みリソースなど

ツール（Tools）の実装

ツールはLLMが何らかのアクションを実行するための機能です。RESTfulAPIのPOSTエンドポイントに似ています。

# 基本的なツールの定義
@mcp.tool()
def calculate_sum(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

# 非同期ツールの定義
@mcp.tool()
async def fetch_data(url: str) -> str:
    """Fetch data from a URL."""
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

ツールの特徴:

• 引数と返り値の型はPythonの型ヒントで指定
• 非同期関数としても実装可能
• コンテキスト（ctx: Context）を受け取って進捗報告なども可能

プロンプト（Prompts）の実装

プロンプトはLLMとの対話パターンを定義するテンプレートです。

# シンプルなプロンプトテンプレート
@mcp.prompt()
def echo_prompt(message: str) -> str:
    """Create a prompt that includes the provided message."""
    return f"Please process this message: {message}"

# 複数メッセージからなるプロンプト
from mcp.types import UserMessage, AssistantMessage

@mcp.prompt()
def debug_error(error: str) -> list[Message]:
    return [
        UserMessage("I'm seeing this error:"),
        UserMessage(error),
        AssistantMessage("I'll help debug that. What have you tried so far?")
    ]

プロンプトの特徴:

• シンプルな文字列または複数のメッセージから構成可能
• ユーザーが選択して呼び出せるテンプレートとして機能

6. 実践的なMCPサーバの例

より実用的なMCPサーバの例を紹介します。以下はWebサイトの内容を取得するサーバです。

"""
Web Fetcher MCP Server
This example demonstrates an MCP server that can fetch and return content from websites.
"""
import httpx
from mcp.server.fastmcp import FastMCP

# MCPサーバを作成
mcp = FastMCP("Web Fetcher")

# Webサイトのコンテンツを取得するツール
@mcp.tool()
async def fetch_website(url: str) -> str:
    """
    Fetch the content from a specified URL.
    """
    # ユーザエージェントを設定（ブロック回避）
    headers = {
        "User-Agent": "MCP Web Fetcher Example"
    }
    
    # httpxで非同期HTTPリクエスト
    async with httpx.AsyncClient(follow_redirects=True, headers=headers) as client:
        try:
            # 指定URLにGETリクエスト送信
            response = await client.get(url)
            
            # エラーコードの場合は例外を発生
            response.raise_for_status()
            
            # レスポンスのテキストを返す
            return response.text
        except httpx.HTTPStatusError as e:
            return f"Error: HTTP status error - {e.response.status_code}"
        except httpx.RequestError as e:
            return f"Error: Request failed - {str(e)}"
        except Exception as e:
            return f"Error: {str(e)}"

# 最近取得したサイトにアクセスするためのリソース
@mcp.resource("webhistory://last")
def last_fetch() -> str:
    """Return information about the last fetched website."""
    # 実際の実装では、取得履歴を追跡
    # これは簡略化した例
    return "No fetch history available yet."

# スクリプト直接実行時にサーバを実行
if __name__ == "__main__":
    # デフォルトでstdioトランスポートを使用
    mcp.run()

データベース連携の例

SQLiteデータベースと連携するMCPサーバの例です：

"""
SQLite Explorer MCP Server
This example demonstrates an MCP server that provides access to a SQLite database.
"""
import sqlite3
from mcp.server.fastmcp import FastMCP, Context

# MCPサーバを作成
mcp = FastMCP("SQLite Explorer")

# このサンプルではインメモリデータベースを使用
DB_PATH = ":memory:"

# サンプルデータでデータベースを初期化
def init_database():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    # usersテーブルを作成
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT UNIQUE NOT NULL,
        age INTEGER
    )
    """)
    
    # サンプルデータを挿入
    sample_users = [
        (1, "Alice Smith", "[e-mail]", 28),
        (2, "Bob Johnson", "[e-mail]", 35),
        (3, "Carol Williams", "[e-mail]", 42),
    ]
    
    # 重複を避けるためINSERT OR REPLACEを使用
    cursor.executemany("""
    INSERT OR REPLACE INTO users (id, name, email, age) VALUES (?, ?, ?, ?)
    """, sample_users)
    
    conn.commit()
    conn.close()

# データベースを初期化
init_database()

# データベーススキーマを取得するリソース
@mcp.resource("schema://tables")
def get_schema() -> str:
    """Retrieve the database schema for all tables."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    # すべてのテーブルを取得
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = cursor.fetchall()
    
    schema = []
    for table in tables:
        table_name = table[0]
        cursor.execute(f"PRAGMA table_info({table_name})")
        columns = cursor.fetchall()
        
        column_info = [f"  {col[1]} {col[2]}" + (" PRIMARY KEY" if col[5] else "") for col in columns]
        schema.append(f"Table: {table_name}\n" + "\n".join(column_info))
    
    conn.close()
    return "\n\n".join(schema)

# データベースにクエリを実行するツール
@mcp.tool()
def query_database(sql: str, ctx: Context) -> str:
    """
    Execute an SQL query against the database.
    """
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    
    try:
        # 実行されるクエリをログに記録
        ctx.info(f"Executing query: {sql}")
        
        # クエリを実行
        cursor.execute(sql)
        
        # カラム名を取得
        column_names = [description[0] for description in cursor.description] if cursor.description else []
        
        # 結果を取得
        results = cursor.fetchall()
        
        # 結果をテーブル形式でフォーマット
        if not column_names or not results:
            return "Query executed successfully. No results or empty result set."
        
        # シンプルなテキストテーブルを構築
        header = " | ".join(column_names)
        separator = "-" * len(header)
        rows = [" | ".join([str(cell) for cell in row]) for row in results]
        
        return f"{header}\n{separator}\n" + "\n".join(rows)
    except Exception as e:
        return f"Error executing query: {str(e)}"
    finally:
        conn.close()

# スクリプト直接実行時にサーバを実行
if __name__ == "__main__":
    mcp.run()

7. クライアントとの連携

MCPサーバを作成したら、クライアントからどのように接続するかを見てみましょう。以下はPythonでMCPクライアントを実装した例です。

"""
Simple MCP Client Example
This example demonstrates how to use the MCP client to connect to and interact with an MCP server.
"""
import asyncio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    # stdio経由でサーバに接続するためのパラメータを定義
    server_params = StdioServerParameters(
        command="python",  # サーバを実行するコマンド
        args=["simple_echo_server.py"],  # MCPサーバを実装するスクリプト
        env=None  # 環境変数（Noneは現在のプロセスから継承）
    )
    
    # サーバに接続
    async with stdio_client(server_params) as (read, write):
        # クライアントセッションを作成
        async with ClientSession(read, write) as session:
            # サーバとの接続を初期化
            await session.initialize()
            
            print("サーバに正常に接続しました！")
            
            # 利用可能なツールをリスト
            tools = await session.list_tools()
            print(f"利用可能なツール: {tools}")
            
            # echoツールを呼び出し
            echo_result = await session.call_tool("echo", {"text": "MCPクライアントからこんにちは！"})
            print(f"Echoツールの結果: {echo_result}")
            
            # 利用可能なリソースをリスト
            resources = await session.list_resources()
            print(f"利用可能なリソース: {resources}")
            
            # 静的リソースを読み込み
            static_content, mime_type = await session.read_resource("echo://static")
            print(f"静的リソースの内容: {static_content}")
            print(f"静的リソースのMIMEタイプ: {mime_type}")
            
            # 動的リソースを読み込み
            dynamic_content, mime_type = await session.read_resource("echo://リソースからこんにちは！")
            print(f"動的リソースの内容: {dynamic_content}")
            print(f"動的リソースのMIMEタイプ: {mime_type}")
            
            # 利用可能なプロンプトをリスト
            prompts = await session.list_prompts()
            print(f"利用可能なプロンプト: {prompts}")
            
            # プロンプトを取得
            if prompts:
                prompt_name = prompts[0].name
                prompt_result = await session.get_prompt(prompt_name, {"message": "このメッセージを処理してください"})
                print(f"プロンプトの結果: {prompt_result}")

if __name__ == "__main__":
    # 非同期メイン関数を実行
    asyncio.run(main())

Claude Desktopとの連携

Claude DesktopでMCPサーバを利用するには、MCPコマンドラインツールを使ってインストールします：

# MCPサーバをClaude Desktopにインストール
mcp install server.py

# カスタム名を指定
mcp install server.py --name "My Analytics Server"

# 環境変数を設定
mcp install server.py -v API_KEY=abc123 -v DB_URL=postgres://...
mcp install server.py -f .env

インストールすると、Claude Desktopの「Servers」セクションにあなたのサーバが表示され、Claude AIがあなたのサーバの機能にアクセスできるようになります。

8. デプロイと運用

MCPサーバのデプロイ方法はいくつかあります：

1. ローカルインストール

開発時や個人利用の場合、mcp installコマンドでClaude Desktopに直接インストールするのが最も簡単です。

2. トランスポートの選択

MCPは複数のトランスポート方式をサポートしています：

• stdio: デフォルトのトランスポート。標準入出力を使用。
• SSE: Server-Sent Eventsを使用したHTTPベースのトランスポート。

SSEトランスポートを使う例：

if __name__ == "__main__":
    import sys
    
    # コマンドライン引数でトランスポートを指定
    transport = "sse" if "--sse" in sys.argv else "stdio"
    port = 8000  # SSEのデフォルトポート
    
    # トランスポートを指定してサーバを実行
    mcp.run(transport=transport, port=port)

3. パッケージング

複数のユーザーで共有する場合は、MCPサーバをPythonパッケージとしてパッケージ化すると便利です：

# pyproject.toml
[build-system]
requires = ["setuptools>=42", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "my-mcp-server"
version = "0.1.0"
description = "My MCP Server"
authors = [{name = "Your Name", email = "[e-mail]"}]
dependencies = [
    "mcp>=1.0.0",
    "httpx>=0.24.0",
]

[project.scripts]
my-mcp-server = "my_mcp_server:main"

4. セキュリティ考慮事項

MCPサーバのデプロイ時には以下のセキュリティ事項を考慮してください：

• リソースへのアクセス範囲を適切に制限する
• 機密情報を扱う場合は適切な認証を実装する
• ユーザー入力を適切にサニタイズする（特にSQLクエリなど）
• 必要最小限の権限でサーバを実行する

9. まとめと次のステップ

本記事では、Model Context Protocol（MCP）の基本概念と、PythonでのMCPサーバ開発方法を紹介しました。MCPを活用することで、AIアプリケーションと外部データソースやツールとの連携が標準化され、開発効率が向上します。

今後の学習に役立つリソース

• Model Context Protocol 公式ドキュメント
• Python SDK リポジトリ
• MCP Servers リポジトリ - 実践的なサーバの例

MCPは急速に発展している技術であり、今後もさまざまなインテグレーションやユースケースが登場することが期待されます。ぜひこの記事を出発点に、あなた自身のMCPサーバを開発して、LLMの可能性を広げてみてください。

参考リンク

• Model Context Protocol 公式サイト
• Python SDK GitHub
• Anthropic Claude
• MCP仕様書