FastMCP 入門 (2) - FastMCPサーバ

「FastMCP」の「サーバ」についてまとめました。

前回

1. FastMCPサーバ

「FastMCPサーバ」は、FastMCPアプリケーションの中心となるクラスで、アプリケーションのツール、リソース、プロンプトのメインコンテナとして機能し、「MCPクライアント」との通信を管理します。

2. サーバの作成

サーバのインスタンスの作成手順は、次のとおりです。
通常は、クライアントアプリケーションやログでサーバを識別できるように、サーバに名前を付けます。

from fastmcp import FastMCP

# 基本的なサーバインスタンスを作成
mcp = FastMCP(name="MyAssistantServer")

# サーバとのやり取り方法に関する指示を追加することもできる
mcp_with_instructions = FastMCP(
    name="HelpfulAssistant",
    instructions="このサーバはデータ分析ツールを提供します。数値データを分析するには、get_average() を呼び出します。"
)

FastMCPのコンストラクタの引数は、次のとおりです。

3. コンポーネント

「FastMCPサーバ」は、いくつかの種類のコンポーネントをクライアントに公開します。

3-1. ツール

「ツール」は、クライアントがアクションを実行したり外部システムにアクセスしたりするために呼び出すことができる関数です。

@mcp.tool()
def multiply(a: float, b: float) -> float:
    """2つの数値を掛け合わせる"""
    return a * b

詳しくはドキュメント「ツール」を参照してください。

3-2. リソース

「リソース」は、クライアントが読み取ることができるデータソースを公開します。

@mcp.resource("data://config")
def get_config() -> dict:
    """アプリケーション構成を提供"""
    return {"theme": "dark", "version": "1.0"}

詳しくはドキュメント「リソースとテンプレート」を参照してください。

3-3. リソーステンプレート

「リソーステンプレート」は、クライアントが特定のデータを要求できるようにするパラメータ化されたリソースです。

@mcp.resource("users://{user_id}/profile")
def get_user_profile(user_id: int) -> dict:
    """IDでユーザープロファイルを取得"""
    # URI内の {user_id} が抽出され、この関数に渡される
    return {"id": user_id, "name": f"User {user_id}", "status": "active"}

詳しくはドキュメント「リソースとテンプレート」を参照してください。

3-4. プロンプト

「プロンプト」は、LLM をガイドするための再利用可能なメッセージテンプレートです。

@mcp.prompt()
def analyze_data(data_points: list[float]) -> str:
    """数値データの分析を求めるプロンプトを作成"""
    formatted_data = ", ".join(str(point) for point in data_points)
    return f"データポイントを分析してください: {formatted_data}"

詳しくはドキュメント「プロンプト」を参照してください。

4. サーバの実行

「FastMCPサーバ」は、クライアントと通信するためにトランスポートメカニズムを必要とします。MCPプロトコルでは、サーバは通常、クライアントが接続する個別のプロセスとして実行されます。

4-1. __main__ ブロックパターン

サーバを実行可能にする標準的な方法は、if __name__ == "__main__": ブロック内に run() を含めることです。

# my_server.py
from fastmcp import FastMCP

mcp = FastMCP(name="MyServer")

@mcp.tool()
def greet(name: str) -> str:
    """ユーザーの名前であいさつ"""
    return f"Hello, {name}!"

if __name__ == "__main__":
    # このコードはファイルが直接実行された場合にのみ実行される
    
    # デフォルト設定での基本実行（stdioトランスポート）
    mcp.run()
    
    # 特定のトランスポートとパラメータの使用
    # mcp.run(transport="sse", host="127.0.0.1", port=9000)

このパターンが重要な理由は、次のとおりです。

「FastMCP」の「CLI」を使用する場合、このパターンは技術的にはオプションですが、すべての「MCPクライアント」との互換性を最大限に高めるためのベストプラクティスと考えられています。

4-2. トランスポートオプション

「FastMCP」は2つのトランスポートメカニズムをサポートしています。

・STDIO トランスポート (デフォルト)
標準入出力 (STDIO) トランスポートは、デフォルトであり、最も互換性の高いオプションです。

# stdioで実行 (デフォルト)
mcp.run()  # または明示的に: mcp.run(transport="stdio")

・SSEトランスポート (Server-Sent Events)
複数のクライアントにサービスを提供する長時間稼働サーバ向けに、FastMCPはSSEをサポートしています。

# デフォルトのホスト/ポート (0.0.0.0:8000) でSSEを使用して実行
mcp.run(transport="sse")

サーバの実行時に、トランスポートパラメータを直接設定できます。

# 特定のパラメータで設定
mcp.run(
    transport="sse", 
    host="127.0.0.1",  # Override default host
    port=8888,         # Override default port
    log_level="debug"  # Set logging level
)

# 同じパラメータで非同期的に実行
import asyncio
asyncio.run(
    mcp.run_sse_async(
        host="127.0.0.1", 
        port=8888, 
        log_level="debug"
    )
)

run() または run_sse_async() に渡されるトランスポートパラメータは、FastMCP インスタンスの作成時に定義された設定を上書きします。SSEトランスポートで最も一般的なパラメータは以下のとおりです。

・高度なトランスポート構成
FastMCP の run() は、内部的には、トランスポート固有の実行メソッドに渡される任意のキーワード引数 (**transport_kwargs) を受け入れます。

# SSEトランスポートの場合、kwargsはrun_sse_async()に渡される
mcp.run(transport="sse", **transport_kwargs)

# stdioトランスポートの場合、kwargsはrun_stdio_async()に渡される
mcp.run(transport="stdio", **transport_kwargs)

将来的に追加されるトランスポート固有のオプションは、コードを変更することなく、同じインターフェースから自動的に利用できるようになります。

・FastMCP CLIの使用
FastMCP CLI は、サーバを実行するための便利な方法を提供します。

# サーバを実行 (デフォルトはstdioトランスポート)
fastmcp run my_server.py:mcp

# トランスポートを明示的に指定
fastmcp run my_server.py:mcp --transport sse

# ホストとポートでSSEトランスポートを構成
fastmcp run my_server.py:mcp --transport sse --host 127.0.0.1 --port 8888

# ログレベル付き
fastmcp run my_server.py:mcp --transport sse --log-level DEBUG

CLI はファイル内の FastMCP サーバ オブジェクトを動的に検索して実行できますが、if __name__ == "__main__": ブロックを含めると、すべてのクライアントとの互換性が確保されます。

5. コンポージングサーバ

FastMCPは、import_server (静的コピー) とmount (ライブリンク) を使用して複数のサーバを統合する機能をサポートしています。これにより、大規模なアプリケーションをモジュール型コンポーネントにまとめたり、既存のサーバを再利用したりすることが可能になります。

詳細、ベストプラクティス、および例については、サーバ構成ガイドを参照してください。

# 例: サブサーバのインポート
from fastmcp import FastMCP
import asyncio

main = FastMCP(name="Main")
sub = FastMCP(name="Sub")

@sub.tool()
def hello(): 
    return "hi"

main.mount("sub", sub)

6. プロキシサーバ

FastMCPは、FastMCP.from_clientを使用することで、任意のMCPサーバ (ローカルまたはリモート) のプロキシとして動作し、トランスポートをブリッジしたり、既存のサーバにフロントエンドを追加したりできます。例えば、リモートSSEサーバをstdio経由でローカルに公開したり、その逆を行ったりできます。

詳しくはプロキシサーバガイドを参照してください。

from fastmcp import FastMCP, Client

backend = Client("http://example.com/mcp/sse")
proxy = FastMCP.from_client(backend, name="ProxyServer")
# 他のFastMCPサーバと同様にプロキシを使用

7. サーバの構成

トランスポート設定 (SSEのhost、port) や重複コンポーネントの処理方法といったサーバの動作は、ServerSettingsで設定できます。これらの設定は、FastMCPの初期化時に渡すか、環境変数（FASTMCP_SERVER_で始まる）で設定するか、.envから読み込むことができます。

from fastmcp import FastMCP

# 初期化中に設定
mcp = FastMCP(
    name="ConfiguredServer",
    port=8080, # ServerSettingsに直接マップ
    on_duplicate_tools="error" # 重複処理を設定
)

# 設定はmcp.settingsからアクセスできる
print(mcp.settings.port) # 出力: 8080
print(mcp.settings.on_duplicate_tools) # 出力: "error"

主要な設定オプションは、次のとおりです。

これらはすべて、FastMCPインスタンスの作成時にパラメータとして直接設定できます。

8. 認証

FastMCP は、MCP プロトコルから OAuth 2.0 認証のサポートを継承し、サーバが認証の背後でツールとリソースを保護できるようにします。

8-1. OAuth 2.0 サポート

mcp.server.auth モジュールは、OAuthServerProvider プロトコルの実装を提供することで、サーバが使用できる OAuth 2.0 サーバインターフェースを実装します。

from fastmcp import FastMCP
from mcp.server.auth.settings import (
    RevocationOptions,
    ClientRegistrationOptions,
    AuthSettings,
)


# 認証付きサーバを作成
mcp = FastMCP(
    name="SecureApp",
    auth_provider=MyOAuthServerProvider(),
    auth=AuthSettings(
        issuer_url="https://myapp.com",
        revocation_options=RevocationOptions(
            enabled=True,
        ),
        client_registration_options=ClientRegistrationOptions(
            enabled=True,
            valid_scopes=["myscope", "myotherscope"],
            default_scopes=["myscope"],
        ),
        required_scopes=["myscope"],
    ),
)

次回