fastmcp-library

FastMCPライブラリ使用ガイド

FastMCPの基本構造

FastMCPは、Model Context Protocol (MCP) サーバーとクライアントを構築するためのPythonフレームワークです。

基本的なサーバー設定

from fastmcp import FastMCP

mcp = FastMCP("不動産情報ライブラリMCPサーバー")

@mcp.tool
def greet(name: str) -> str:
    """指定された名前で挨拶を返します。"""
    return f"こんにちは、{name}さん！"

if __name__ == "__main__":
    mcp.run(transport="stdio")  # stdio, http, sse をサポート

ツール定義パターン

1. 基本的なツール定義

@mcp.tool
def add(a: int, b: int) -> int:
    """2つの整数を加算します。"""
    return a + b

2. カスタム名とメタデータ付きツール

@mcp.tool(
    name="find_products",
    description="製品カタログを検索します",
    tags={"catalog", "search"},
    meta={"version": "1.2", "author": "team"}
)
def search_products(query: str, category: str | None = None) -> list[dict]:
    """製品検索の実装"""
    return [{"id": 1, "name": "商品名"}]

3. ツールアノテーション

from mcp.types import ToolAnnotations

@mcp.tool(
    annotations=ToolAnnotations(
        title="計算ツール",
        readOnlyHint=True,      # 読み取り専用
        destructiveHint=False,  # 破壊的でない
        idempotentHint=True,    # 冪等性あり
        openWorldHint=False     # 閉じた世界
    )
)
def calculate_sum(a: float, b: float) -> float:
    """数値の合計を計算します。"""
    return a + b

リソース定義

@mcp.resource("system://status")
def get_system_status() -> dict:
    """システムステータスを返します。"""
    return {"status": "正常稼働中"}

@mcp.resource("weather://{city}")
def get_weather(city: str) -> str:
    """指定都市の天気を取得します。"""
    return f"{city}の天気情報"

プロンプト定義

@mcp.prompt
def ask_about_topic(topic: str) -> str:
    """トピックについて説明を求めるプロンプトを生成します。"""
    return f"'{topic}'について詳しく説明してください。"

@mcp.prompt(
    name="analyze_request",
    description="データ分析リクエストを生成",
    tags={"analysis", "data"}
)
def data_analysis_prompt(data_uri: str, analysis_type: str = "summary") -> str:
    """データ分析のプロンプトを生成します。"""
    return f"{data_uri}のデータに対して{analysis_type}分析を実行してください。"

コンテキストの使用

from fastmcp import Context

@mcp.tool
async def process_data(uri: str, ctx: Context) -> str:
    """コンテキストを使用したデータ処理。"""
    # ログ出力
    await ctx.info(f"処理開始: {uri}")
    await ctx.debug("デバッグ情報")
    
    # プログレス報告
    await ctx.report_progress(50, 100, "処理中...")
    
    # リソース読み取り
    data = await ctx.read_resource(uri)
    
    # 状態管理
    ctx.set_state("processing", True)
    
    return f"処理完了: {uri}"