FastMCP 入門 (4) - リソース

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"],
    }

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

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
    }

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 "ログファイルが見つかりません。"

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リストを返す

# 標準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"を使用して保存およびアクセス

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
    }

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

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
    }

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": "ルックアップパラメータが指定されていません"}

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}