FastMCP 入門 (4) - リソース

「FastMCP」の「リソース」についてまとめました。

前回

1. リソース

「リソース」はMCPクライアントが読み取れるデータまたはファイルを表します。「リソーステンプレート」はこの概念を拡張し、クライアントがURIで渡されたパラメータに基づいて動的に生成されたリソースをリクエストできるようにします。

FastMCPは、主に@mcp.resourceデコレータを使用して、静的リソースと動的リソースの両方の定義を簡素化します。

これにより、LLMはファイル、データベースコンテンツ、構成、会話に関連する動的生成された情報にアクセスできます。

2. @mcp.resourceデコレータ

リソースを定義する最も一般的な方法は、Python関数をデコレートすることです。デコレータには、リソースの一意のURIが必要です。

import json
from fastmcp import FastMCP

mcp = FastMCP(name="DataServer")

# 文字列を返す基本的な動的リソース
@mcp.resource("resource://greeting")
def get_greeting() -> str:
    """簡単な挨拶メッセージを提供"""
    return "FastMCP Resources からこんにちは!"

# JSON データを返すリソース (dict は自動シリアライズ)
@mcp.resource("data://config")
def get_config() -> dict:
    """アプリケーション構成を JSONとして提供"""
    return {
        "theme": "dark",
        "version": "1.2.0",
        "features": ["tools", "resources"],
    }

3. 戻り値

FastMCPは、関数の戻り値を適切なMCPリソースコンテンツに自動的に変換します。

4. リソースのメタデータ

デコレータの引数を使用して、リソースのプロパティをカスタマイズできます。

from fastmcp import FastMCP

mcp = FastMCP(name="DataServer")

# メタデータを指定する例
@mcp.resource(
    uri="data://app-status",      # 明示的なURI (必須)
    name="ApplicationStatus",     # カスタム名
    description="アプリケーションの現在のステータスを提供", # カスタム説明
    mime_type="application/json", # 明示的なMIMEタイプ
    tags={"monitoring", "status"} # 分類タグ
)
def get_application_status() -> dict:
    """内部関数の説明 (上記に説明が提供されている場合は無視されます)"""
    return {"status": "ok", "uptime": 12345, "version": mcp.settings.version} # Example usage

5. MCPコンテキストへのアクセス

リソースとリソーステンプレートは、Contextオブジェクトを介してMCPの追加情報と機能にアクセスできます。これにアクセスするには、リソース関数にContext型アノテーションを付与したパラメータを追加します。

from fastmcp import FastMCP, Context

mcp = FastMCP(name="DataServer")

@mcp.resource("resource://system-status")
async def get_system_status(ctx: Context) -> dict:
    """システムステータス情報を提供"""
    return {
        "status": "operational",
        "request_id": ctx.request_id
    }

@mcp.resource("resource://{name}/details")
async def get_details(name: str, ctx: Context) -> dict:
    """特定の名前の詳細を取得"""
    return {
        "name": name,
        "accessed_at": ctx.request_id
    }

詳しくは、Contextのドキュメントを参照してください。

6. 非同期リソース

サーバのブロックを回避するため、I/O操作 (データベースやネットワークからの読み取りなど) を実行するリソース関数には async def を使用します。

import aiofiles
from fastmcp import FastMCP

mcp = FastMCP(name="DataServer")

@mcp.resource("file:///app/data/important_log.txt", mime_type="text/plain")
async def read_important_log() -> str:
    """特定のログファイルから非同期的にコンテンツを読み取る"""
    try:
        async with aiofiles.open("/app/data/important_log.txt", mode="r") as f:
            content = await f.read()
        return content
    except FileNotFoundError:
        return "ログファイルが見つかりません。"

7. リソースクラス

@mcp.resource は動的コンテンツに最適ですが、mcp.add_resource() と具体的な Resourceサブクラスを使用して、定義済みのリソース (静的ファイルや単純なテキストなど) を直接登録することもできます。

from pathlib import Path
from fastmcp import FastMCP
from fastmcp.resources import FileResource, TextResource, DirectoryResource

mcp = FastMCP(name="DataServer")

# 1. 静的ファイルを直接公開
readme_path = Path("./README.md").resolve()
if readme_path.exists():
    # Use a file:// URI scheme
    readme_resource = FileResource(
        uri=f"file://{readme_path.as_posix()}",
        path=readme_path, # Path to the actual file
        name="README File",
        description="The project's README.",
        mime_type="text/markdown",
        tags={"documentation"}
    )
    mcp.add_resource(readme_resource)

# 2. シンプルな定義済みテキストを公開
notice_resource = TextResource(
    uri="resource://notice",
    name="Important Notice",
    text="日曜日にシステムメンテナンスが予定されています。",
    tags={"notification"}
)
mcp.add_resource(notice_resource)

# 3. URIとは異なるカスタムキーを使用
special_resource = TextResource(
    uri="resource://common-notice",
    name="Special Notice",
    text="これはカスタムストレージキーに関する特別な通知です。",
)
mcp.add_resource(special_resource, key="resource://custom-key")

# 4. ディレクトリリストの公開
data_dir_path = Path("./app_data").resolve()
if data_dir_path.is_dir():
    data_listing_resource = DirectoryResource(
        uri="resource://data-files",
        path=data_dir_path, # ディレクトリへのパス
        name="Data Directory Listing",
        description="データディレクトリで使用可能なファイルを一覧表示します。",
        recursive=False # サブディレクトリを一覧表示するにはTrueに設定
    )
    mcp.add_resource(data_listing_resource) # ファイルのJSONリストを返す

コンテンツが静的であるか、ファイル/URLから直接取得される場合にこれらを使用して、専用のPython関数の必要性を回避します。

8. カスタムリソースキー

mcp.add_resource() を使用してリソースを直接追加する場合は、オプションでカスタムストレージキーを指定できます。

# 標準URIをキーとしてリソースを作成
resource = TextResource(uri="resource://data")
mcp.add_resource(resource)  # "resource://data" を使用して保存およびアクセス

# カスタムキーを使用してリソースを作成する
special_resource = TextResource(uri="resource://special-data")
mcp.add_resource(special_resource, key="internal://data-v2")  # "internal://data-v2"を使用して保存およびアクセス

このパラメータは、デコレータを使用するときに URI が明示的に提供されるため、add_resource() を直接使用する場合にのみ使用可能であり、@resourceデコレータ経由は使用できません。

9. リソーステンプレート

リソーステンプレートを使用すると、クライアントはURIに埋め込まれたパラメータに応じてコンテンツが変化するリソースをリクエストできます。テンプレートは@mcp.resourceデコレータを使用して定義しますが、URI文字列に{parameter_name}プレースホルダを含め、対応する引数を関数シグネチャに追加します。

リソーステンプレートは、通常のリソースとほとんどの設定オプション（名前、説明、MIMEタイプ、タグ）を共有しますが、関数パラメータにマッピングされるURIパラメータを定義する機能が追加されています。

リソーステンプレートは、パラメータの一意のセットごとに新しいリソースを生成するため、リソースをオンデマンドで動的に作成できます。例えば、リソーステンプレート「user://profile/{name}」が登録されている場合、MCPクライアントは「user://profile/ford」または「user://profile/marvin」をリクエストすることで、これら2つのユーザープロファイルのいずれかをリソースとして取得できます。各リソースを個別に登録する必要はありません。

2つのリソーステンプレートを定義する方法を示す完全な例を以下に示します。

from fastmcp import FastMCP

mcp = FastMCP(name="DataServer")

# Template URI includes {city} placeholder
@mcp.resource("weather://{city}/current")
def get_weather(city: str) -> dict:
    """特定の都市の天気情報を提供"""
    # 実際の実装では、天気予報APIを呼び出す
    # ここでは例として簡略化したロジックを使用
    return {
        "city": city.capitalize(),
        "temperature": 22,
        "condition": "Sunny",
        "unit": "celsius"
    }

# 複数のパラメータを持つテンプレート
@mcp.resource("repos://{owner}/{repo}/info")
def get_repo_info(owner: str, repo: str) -> dict:
    """GitHub リポジトリに関する情報を取得"""
    # 実際の実装では、GitHub APIを呼び出すことになる。
    return {
        "owner": owner,
        "name": repo,
        "full_name": f"{owner}/{repo}",
        "stars": 120,
        "forks": 48
    }

10. ワイルドカードパラメータ

リソーステンプレートは、複数のパスセグメントに一致するワイルドカードパラメータをサポートしています。標準パラメータ（{param}）は単一のパスセグメントにのみ一致し、「/」境界を越えることはありません。一方、ワイルドカードパラメータ（{param*}）は、スラッシュを含む複数のセグメントをキャプチャできます。ワイルドカードは、URIテンプレートの定義部分（リテラルまたは別のパラメータ）までの、後続のすべてのパスセグメントをキャプチャします。これにより、単一のURIテンプレートに複数のワイルドカードパラメータを含めることができます。

from fastmcp import FastMCP

mcp = FastMCP(name="DataServer")


# 標準パラメータは1つのセグメントのみに一致
@mcp.resource("files://{filename}")
def get_file(filename: str) -> str:
    """名前でファイルを取得"""
    # files://<single-segment> のみに一致
    return f"File content for: {filename}"


# ワイルドカードパラメータは複数のセグメントに一致できる
@mcp.resource("path://{filepath*}")
def get_path_content(filepath: str) -> str:
    """特定のパスにあるコンテンツを取得"""
    # path://docs/server/resources.mdx に一致
    return f"Content at path: {filepath}"


# 標準パラメータとワイルドカードパラメータの混在
@mcp.resource("repo://{owner}/{path*}/template.py")
def get_template_file(owner: str, path: str) -> dict:
    """特定のリポジトリとパスからファイルを取得します
    ただし、リソースが `template.py` で終わる場合のみ"""
    # repo://jlowin/fastmcp/src/resources/template.py に一致
    return {
        "owner": owner,
        "path": path + "/template.py",
        "content": f"File at {path}/template.py in {owner}'s repository"
    }

ワイルドカードパラメータは、次のような場合に役立ちます。

通常のパラメータと同様に、ワイルドカードパラメータは関数シグネチャ内で名前付きパラメータとして指定する必要があり、必要なすべての関数パラメータは URI テンプレートに記述する必要があることに注意してください。

11. デフォルト値

リソーステンプレートを作成する際、FastMCP は URI テンプレートパラメータと関数パラメータの関係について 2 つのルールを適用します。

ただし、デフォルト値を持つ関数パラメータは、URIテンプレートに含める必要はありません。クライアントがリソースをリクエストすると、FastMCPは次の処理を行います。

これにより、柔軟なAPI設計が可能になります。例えば、オプションパラメータ付きのシンプルな検索テンプレートは次のようになります。

from fastmcp import FastMCP

mcp = FastMCP(name="DataServer")

@mcp.resource("search://{query}")
def search_resources(query: str, max_results: int = 10, include_archived: bool = False) -> dict:
    """クエリ文字列に一致するリソースを検索"""
    # URIでは「クエリ」のみが必要で、他のパラメータはデフォルトを使用
    results = perform_search(query, limit=max_results, archived=include_archived)
    return {
        "query": query,
        "max_results": max_results,
        "include_archived": include_archived,
        "results": results
    }

このテンプレートを使用すると、クライアントは search://python をリクエストでき、query="python"、max_results=10、include_archived=False で関数が呼び出されます。MCP 開発者は、より具体的なパラメータを指定して、基盤となる search_resources 関数を直接呼び出すこともできます。

さらに強力なパターンとして、1つの関数を複数の URI テンプレートに登録し、同じデータに複数の方法でアクセスできるようにする方法があります。

from fastmcp import FastMCP

mcp = FastMCP(name="DataServer")

# 異なる識別子でアクセスできるユーザー検索関数を定義
@mcp.resource("users://email/{email}")
@mcp.resource("users://name/{name}")
def lookup_user(name: str | None = None, email: str | None = None) -> dict:
    """名前またはメールでユーザーを検索"""
    if email:
        return find_user_by_email(email) # 疑似コード
    elif name:
        return find_user_by_name(name) # 疑似コード
    else:
        return {"error": "ルックアップパラメータが指定されていません"}

LLMまたはクライアントは、以下の2つの方法でユーザー情報を取得できます。

このスタックデコレータパターンでは、次のようになります。

・テンプレートの動作

テンプレートは、RESTのような原則に従ってパラメータ化されたデータアクセスポイントを公開する強力な方法を提供します。

12. サーバの動作

12-1. 重複リソース

FastMCPサーバが同じURIに複数のリソースまたはテンプレートを登録しようとした場合の処理​​方法を設定できます。FastMCPの初期化時にon_duplicate_resources設定を使用してください。

from fastmcp import FastMCP

mcp = FastMCP(
    name="ResourceServer",
    on_duplicate_resources="error" # 重複がある場合はエラーを発生
)

@mcp.resource("data://config")
def get_config_v1(): return {"version": 1}

# "data://config" はすでに登録されており、動作は "error" であるため、
# この登録の試行では ValueError が発生
# @mcp.resource("data://config")
# def get_config_v2(): return {"version": 2}

次回