MCP Python SDK

モデルコンテキストプロトコル (MCP) の Python 実装

概要

モデルコンテキストプロトコル（MCP）は、アプリケーションがLLMに標準化された方法でコンテキストを提供することを可能にし、コンテキスト提供に関する懸念をLLMとの実際のやり取りから分離します。このPython SDKはMCP仕様を完全に実装しており、以下のことが容易になります。

任意のMCPサーバーに接続できるMCPクライアントを構築する
リソース、プロンプト、ツールを公開するMCPサーバーを作成する
stdioやSSEなどの標準トランスポートを使用する
すべてのMCPプロトコルメッセージとライフサイクルイベントを処理する

インストール

Python プロジェクトを管理するには、 uvを使用することをお勧めします。

uv add "mcp[cli]"

あるいは:

pip install mcp

クイックスタート

計算ツールといくつかのデータを公開するシンプルな MCP サーバーを作成しましょう。

# 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

あるいは、MCP Inspector を使用してテストすることもできます。

mcp dev server.py

MCPとは何ですか?

モデルコンテキストプロトコル（MCP）を使用すると、LLMアプリケーションにデータと機能を安全かつ標準化された方法で公開するサーバーを構築できます。LLMとのやり取りに特化したWeb APIのようなものと考えてください。MCPサーバーは以下のことが可能です。

リソースを通じてデータを公開する (これらは GET エンドポイントのようなもので、LLM のコンテキストに情報をロードするために使用されます)
ツールを通じて機能を提供する（POSTエンドポイントのようなもので、コードを実行したり、副作用を発生させたりするために使用されます）
プロンプト（LLMインタラクション用の再利用可能なテンプレート）を通じてインタラクションパターンを定義する
さらにもっと！

コアコンセプト

サーバ

FastMCPサーバーは、MCPプロトコルへのコアインターフェースです。接続管理、プロトコル準拠、メッセージルーティングを処理します。

from mcp.server.fastmcp import FastMCP

# Create a named server
mcp = FastMCP("My App")

# Specify dependencies for deployment and development
mcp = FastMCP("My App", dependencies=["pandas", "numpy"])

リソース

リソースは、LLMにデータを公開する方法です。REST APIのGETエンドポイントに似ています。データを提供しますが、大きな計算を実行したり、副作用を引き起こしたりすることはありません。

@mcp.resource("config://app")
def get_config() -> str:
    """Static configuration data"""
    return "App configuration here"

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """Dynamic user data"""
    return f"Profile data for user {user_id}"

ツール

ツールは、LLMがサーバーを介してアクションを実行できるようにします。リソースとは異なり、ツールは計算を実行し、副作用を持つことが期待されます。

@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
    """Calculate BMI given weight in kg and height in meters"""
    return weight_kg / (height_m ** 2)

@mcp.tool()
async def fetch_weather(city: str) -> str:
    """Fetch current weather for a city"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.weather.com/{city}")
        return response.text

プロンプト

プロンプトは、LLM がサーバーと効果的に対話するのに役立つ再利用可能なテンプレートです。

@mcp.prompt()
def review_code(code: str) -> str:
    return f"Please review this code:\n\n{code}"

@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?")
    ]

画像

FastMCP は、画像データを自動的に処理するImageクラスを提供します。

from mcp.server.fastmcp import FastMCP, Image
from PIL import Image as PILImage

@mcp.tool()
def create_thumbnail(image_path: str) -> Image:
    """Create a thumbnail from an image"""
    img = PILImage.open(image_path)
    img.thumbnail((100, 100))
    return Image(data=img.tobytes(), format="png")

コンテクスト

Context オブジェクトは、ツールとリソースに MCP 機能へのアクセスを提供します。

from mcp.server.fastmcp import FastMCP, Context

@mcp.tool()
async def long_task(files: list[str], ctx: Context) -> str:
    """Process multiple files with progress tracking"""
    for i, file in enumerate(files):
        ctx.info(f"Processing {file}")
        await ctx.report_progress(i, len(files))
        data = await ctx.read_resource(f"file://{file}")
    return "Processing complete"

サーバーの実行

開発モード

サーバーをテストおよびデバッグする最も速い方法は、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 にインストールします。

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

直接実行

カスタム展開などの高度なシナリオの場合:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("My App")

if __name__ == "__main__":
    mcp.run()

次のコマンドで実行します:

python server.py
# or
mcp run server.py

例

エコーサーバー

リソース、ツール、プロンプトを示すシンプルなサーバー:

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 エクスプローラー

データベース統合を示すより複雑な例:

from mcp.server.fastmcp import FastMCP
import sqlite3

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)}"

高度な使用法

低レベルサーバー

より詳細な制御が必要な場合は、低レベルのサーバー実装を直接使用できます。これにより、プロトコルへのフルアクセスが可能になり、サーバーのあらゆる側面をカスタマイズできます。

from mcp.server.lowlevel import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types

# Create a server instance
server = Server("example-server")

@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
    return [
        types.Prompt(
            name="example-prompt",
            description="An example prompt template",
            arguments=[
                types.PromptArgument(
                    name="arg1",
                    description="Example argument",
                    required=True
                )
            ]
        )
    ]

@server.get_prompt()
async def handle_get_prompt(
    name: str,
    arguments: dict[str, Any] | None
) -> types.GetPromptResult:
    if name != "example-prompt":
        raise ValueError(f"Unknown prompt: {name}")

    return types.GetPromptResult(
        description="Example prompt",
        messages=[
            types.PromptMessage(
                role="user",
                content=types.TextContent(
                    type="text",
                    text="Example prompt text"
                )
            )
        ]
    )

async def run():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="example",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                )
            )
        )

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

MCPクライアントの作成

SDK は、MCP サーバーに接続するための高レベルのクライアントインターフェイスを提供します。

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

# Create server parameters for stdio connection
server_params = StdioServerParameters(
    command="python", # Executable
    args=["example_server.py"], # Optional command line arguments
    env=None # Optional environment variables
)

async def run():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the connection
            await session.initialize()

            # List available prompts
            prompts = await session.list_prompts()

            # Get a prompt
            prompt = await session.get_prompt("example-prompt", arguments={"arg1": "value"})

            # List available resources
            resources = await session.list_resources()

            # List available tools
            tools = await session.list_tools()

            # Read a resource
            resource = await session.read_resource("file://some/path")

            # Call a tool
            result = await session.call_tool("tool-name", arguments={"arg1": "value"})

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

MCPプリミティブ

MCP プロトコルは、サーバーが実装できる 3 つのコアプリミティブを定義します。

原生的	コントロール	説明	使用例
プロンプト	ユーザー制御	ユーザーの選択によって呼び出されるインタラクティブなテンプレート	スラッシュコマンド、メニューオプション
リソース	アプリケーション制御	クライアントアプリケーションによって管理されるコンテキストデータ	ファイルの内容、API 応答
ツール	モデル制御	LLMに公開されアクションを実行する関数	API呼び出し、データ更新

サーバー機能

MCP サーバーは初期化中に機能を宣言します。

能力	機能フラグ	説明
`prompts`	`listChanged`	プロンプトテンプレート管理
`resources`	`subscribe` `listChanged`	リソースの公開と更新
`tools`	`listChanged`	ツールの発見と実行
`logging`	-	サーバーのログ記録構成
`completion`	-	引数補完の提案

ドキュメント

貢献

私たちは、経験レベルを問わず、貢献者の方々を熱心にサポートし、プロジェクトへのご参加を心よりお待ちしております。まずは貢献ガイドをご覧ください。

ライセンス

このプロジェクトは MIT ライセンスに基づいてライセンスされています - 詳細については LICENSE ファイルを参照してください。

FastMCP

Integrations