FastMCP 入門 (3) - ツール

「FastMCP」の「ツール」についてまとめました。

前回

1. ツール

「ツール」は、LLMが外部システムと連携し、「コードの実行」や「データへのアクセス」を可能にする構成要素です。FastMCPでは、「ツール」はMCPプロトコルを介してLLMに公開されるPython関数になります。

LLMがツールの使用を決定すると、以下の処理が行われます。

これにより、LLMはデータベースのクエリ、APIの呼び出し、計算、ファイルへのアクセスなどのタスクを実行できます。

2. @mcp.tool デコレータ

ツールを作成するには、Python 関数を @mcp.tool() で装飾するだけです。

from fastmcp import FastMCP

mcp = FastMCP(name="CalculatorServer")

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

このツールを登録すると、FastMCPは自動的に以下の処理を行います。

3. パラメータ

3-1. パラメータの型アノテーション

パラメータの型アノテーションは、ツールが適切に機能するために不可欠です。型アノテーションには、以下の機能があります。

@mcp.tool()
def analyze_text(
    text: str,
    max_tokens: int = 100,
    language: str | None = None
) -> dict:
    """提供されたテキストを分析"""
    # 実装...

3-2. パラメータのメタデータ

PydanticのFieldとAnnotatedを使用することで、パラメータのメタデータを提供できます。型ヒントと検証ルールを分離できるため、推奨されます。

from typing import Annotated
from pydantic import Field

@mcp.tool()
def process_image(
    image_url: Annotated[str, Field(description="処理する画像のURL")],
    resize: Annotated[bool, Field(description="画像のサイズを変更するかどうか")] = False,
    width: Annotated[int, Field(description="ターゲットの幅 (ピクセル)", ge=1, le=2000)] = 800,
    format: Annotated[
        Literal["jpeg", "png", "webp"], 
        Field(description="出力画像形式")
    ] = "jpeg"
) -> dict:
    """オプションでサイズ変更して画像を処理"""
    # 実装...

Fieldをデフォルト値として使用することもできますが、Annotatedのアプローチが推奨されます。

@mcp.tool()
def search_database(
    query: str = Field(description="検索クエリ文字列"),
    limit: int = Field(10, description="結果の最大数", ge=1, le=100)
) -> list:
    """提供されたクエリを使用してデータベースを検索"""
    # 実装...

Fieldには、いくつかの検証機能とドキュメント機能があります。

3-3. サポートされる型

FastMCPは、すべての Pydantic型を含む幅広い型アノテーションをサポートしています。

3-4. オプション引数

FastMCPは Pythonの標準的な関数パラメータ規則に従います。デフォルト値のないパラメータは必須ですが、デフォルト値のあるパラメータはオプションです。

@mcp.tool()
def search_products(
    query: str,                   # 必須 - デフォルト値なし
    max_results: int = 10,        # オプション - デフォルト値あり
    sort_by: str = "relevance",   # オプション - デフォルト値あり
    category: str | None = None   # オプション - None も指定可能
) -> list[dict]:
    """製品カタログを検索"""
    # 実装...

この例では、LLMはクエリ パラメータを提供する必要がありますが、max_results、sort_by、category は明示的に提供されない場合、デフォルト値を使用します。

4. メタデータ

FastMCPは関数から名前と説明を推測しますが、@mcp.tool デコレータに引数を指定してこれらをオーバーライドすることもできます。

@mcp.tool(
    name="find_products",  # ツール名
    description="オプションのカテゴリフィルタリングを使用して製品カタログを検索",  # 説明
    tags={"catalog", "search"}  # 整理/フィルタリングのためのオプションのタグ
)
def search_products_implementation(query: str, category: str | None = None) -> list[dict]:
    """内部関数の説明 (上記に説明が指定されている場合は無視される) """
    # 実装...
    print(f"Searching for '{query}' in category '{category}'")
    return [{"id": 2, "name": "Another Product"}]

5. 非同期ツール

FastMCPは、標準関数 (def) と非同期関数 (async def) の両方をツールとしてシームレスにサポートします。

# 同期ツール (CPUバウンドまたはクイックタスクに適している)
@mcp.tool()
def calculate_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """2つの座標間の距離を計算"""
    # 実装...
    return 42.5

# 非同期ツール (I/Oバウンド操作に最適)
@mcp.tool()
async def fetch_weather(city: str) -> dict:
    """都市の現在の気象状況を取得"""
    # ネットワーク呼び出しやファイルI/Oなどを含む操作には async def を使用
    # 外部操作を待機している間にサーバがブロックされるのを防ぐ
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/weather/{city}") as response:
            # 戻る前に応答ステータスを確認
            response.raise_for_status()
            return await response.json()

サーバの応答性を維持するために、ツールが外部システム (ネットワーク リクエスト、データベース クエリ、ファイル アクセス) を待機する可能性のある操作を実行する場合は、async def を使用します。

6. 戻り値

FastMCPは、関数から返された値をクライアントに適したMCPコンテンツ形式に自動変換します。

FastMCPは可能であれば、他の型を文字列にシリアル化しようとします。

from fastmcp import FastMCP, Image
import io
try:
    from PIL import Image as PILImage
except ImportError:
    raise ImportError("この例を実行するには、pillow ライブラリをインストールしてください。")

mcp = FastMCP("Image Demo")

@mcp.tool()
def generate_image(width: int, height: int, color: str) -> Image:
    """単色画像を生成"""
    # Pillowを使って画像を作成
    img = PILImage.new("RGB", (width, height), color=color)

    # バイトバッファに保存
    buffer = io.BytesIO()
    img.save(buffer, format="PNG")
    img_bytes = buffer.getvalue()

    # FastMCPのイメージヘルパーを使用して返す
    return Image(data=img_bytes, format="png")

@mcp.tool()
def do_nothing() -> None:
    """このツールはアクションを実行しますが、データを返しません"""
    print("サイドエフェクトを実行...")
    return None

7. エラー処理

ツールでエラーが発生した場合は、標準のPython例外（ValueError、TypeError、FileNotFoundError、カスタム例外など）を発生させるだけです。

@mcp.tool()
def divide(a: float, b: float) -> float:
    """aをbで割る"""
    if b == 0:
        # 例外を発生させる
        raise ValueError("ゼロによる除算は許可されていません。")
    if not isinstance(a, (int, float)) or not isinstance(b, (int, float)):
        raise TypeError("両方の引数は数値でなければなりません。")
    return a / b

FastMCPは、ツール関数内で発生した例外を自動的にキャッチします。

8. アノテーション

FastMCPを使用すると、アノテーションを通じてツールに特殊なメタデータを追加できます。これらのアノテーションは、LLMプロンプトのトークンコンテキストを消費することなく、ツールの動作をクライアントアプリケーションに伝えます。

クライアントアプリケーションにおけるアノテーションの目的はいくつかあります。

@mcp.tool()デコレータのannotations パラメータを使用して、ツールにアノテーションを追加できます。

@mcp.tool(
    annotations={
        "title": "Calculate Sum",
        "readOnlyHint": True,
        "openWorldHint": False
    }
)
def calculate_sum(a: float, b: float) -> float:
    """2つの数字を足し合わせる"""
    return a + b

FastMCPは次の標準アノテーションをサポートしています。

アノテーションはユーザーエクスペリエンスの向上に役立ちますが、あくまでもアドバイス的なヒントとして扱うべきです。クライアントアプリケーションが適切なUI要素や安全対策を提示するのに役立ちますが、それ自体でセキュリティ境界を強制するものではありません。

9. MCP コンテキスト

ツールは、コンテキストオブジェクトを介して、ログ記録、リソースの読み取り、進行状況の報告などのMCP機能にアクセスできます。コンテキストを使用するには、ツール関数に型ヒント Context を指定したパラメータを追加します。

from fastmcp import FastMCP, Context

mcp = FastMCP(name="ContextDemo")

@mcp.tool()
async def process_data(data_uri: str, ctx: Context) -> dict:
    """進捗状況レポートを使用してリソースからのデータを処理"""
    await ctx.info(f"Processing data from {data_uri}")
    
    # リソースの読み込み
    resource = await ctx.read_resource(data_uri)
    data = resource[0].content if resource else ""
    
    # 進捗状況の報告
    await ctx.report_progress(progress=50, total=100)
    
    # クライアントのLLMへのサポート依頼例
    summary = await ctx.sample(f"Summarize this in 10 words: {data[:200]}")
    
    await ctx.report_progress(progress=100, total=100)
    return {
        "length": len(data),
        "summary": summary.text
    }

Contextオブジェクトは、以下のアクセスを提供します。

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

10. パラメータ型

FastMCPは、ツール設計の柔軟性を高めるため、幅広いパラメータ型をサポートしています。

FastMCPは、Pydantic がフィールドとしてサポートするすべての型 (Pydantic のすべてのカスタム型を含む) を基本的にサポートしています。つまり、Pydanticで検証および解析できるあらゆる型をツールパラメータで使用できます。

FastMCPは、可能な場合は型強制をサポートします。つまり、クライアントが想定される型と一致しないデータを送信した場合、FastMCPはそれを適切な型に変換しようとします。例えば、クライアントが int としてアノテーションされたパラメータに文字列を送信した場合、FastMCPはそれを整数に変換しようとします。変換できない場合、FastMCPは検証エラーを返します。

10-1. 組み込み型

最も一般的なパラメータ型は、Pythonの組み込みスカラー型です。

@mcp.tool()
def process_values(
    name: str,             # テキストデータ
    count: int,            # 整数
    amount: float,         # 浮動小数点数
    enabled: bool          # ブール値 (True/False)
):
    """さまざまな値のタイプを処理"""
    # 実装...

これらの型は、LLMに許容される値に関する明確な期待値を提供し、FastMCPが入力を適切に検証できるようにします。クライアントが「42」のような文字列を提供した場合でも、int型でアノテーションされたパラメータの場合は整数に変換されます。

10-2. 日付と時刻の型

FastMCPは、datetimeモジュールのさまざまな日付と時刻の型をサポートしています。

from datetime import datetime, date, timedelta

@mcp.tool()
def process_date_time(
    event_date: date,             # ISO形式の日付文字列または日付オブジェクト
    event_time: datetime,         # ISO形式の日付時刻文字列または日付時刻オブジェクト
    duration: timedelta = timedelta(hours=1)  # 整数秒またはtimedelta
) -> str:
    """日付と時刻情報を処理"""
    # 型は文字列から自動的に変換される
    assert isinstance(event_date, date)  
    assert isinstance(event_time, datetime)
    assert isinstance(duration, timedelta)
    
    return f"Event on {event_date} at {event_time} for {duration}"

10-3. コレクション型

FastMCPは、すべての標準的なPythonコレクション型をサポートしています。

@mcp.tool()
def analyze_data(
    values: list[float],           # 数値のリスト
    properties: dict[str, str],    # 文字列のキーと値を持つ辞書
    unique_ids: set[int],          # 一意の整数の集合
    coordinates: tuple[float, float],  # 固定構造のタプル
    mixed_data: dict[str, list[int]] # ネストされたコレクション
):
    """データのコレクションを分析"""
    # 実装...

すべてのコレクション型はパラメータアノテーションとして使用できます。

コレクション型はネストしたり組み合わせたりすることで、複雑なデータ構造を表すことができます。想定される構造に一致するJSON文字列は自動的に解析され、適切なPythonコレクション型に変換されます。

10-4. Union型とOptional型

複数の型を受け入れることができる、または省略可能なパラメータの場合

@mcp.tool()
def flexible_search(
    query: str | int,              # 文字列または整数のいずれか
    filters: dict[str, str] | None = None,  # 辞書 (オプション)
    sort_field: str | None = None  # 文字列 (オプション)
):
    """柔軟なパラメータタイプで検索"""
    # 実装...

最新のPython構文 (str | int) は、古い Union[str, int] 形式よりも優先されます。同様に、 str | None は Optional[str] よりも優先されます。

10-5. 制約型

パラメータが定義済みの値セットのいずれかである必要がある場合、リテラル型または列挙型のいずれかを使用できます。

・リテラル
リテラルはパラメータを特定の値のセットに制限します。

from typing import Literal

@mcp.tool()
def sort_data(
    data: list[float],
    order: Literal["ascending", "descending"] = "ascending",
    algorithm: Literal["quicksort", "mergesort", "heapsort"] = "quicksort"
):
    """特定のオプションを使用してデータを並べ替える"""
    # 実装...

・列挙型
制約された値のより構造化されたセットには、PythonのEnumクラスを使用します。

from enum import Enum

class Color(Enum):
    RED = "red"
    GREEN = "green"
    BLUE = "blue"

@mcp.tool()
def process_image(
    image_path: str, 
    color_filter: Color = Color.RED
):
    """カラーフィルターを使用して画像を処理"""
    # 実装...
    # color_filterはColor列挙型のメンバーになる

10-6. バイナリデータ

ツールパラメータでバイナリデータを処理するには、次の2つの方法があります。

・Bytes

@mcp.tool()
def process_binary(data: bytes):
    """バイナリデータを直接処理します。

    クライアントはバイナリ文字列を送信でき、それは直接バイトに変換されます。
    """
    # バイナリデータを使用した実装
    data_length = len(data)
    # ...

パラメータをバイト列としてアノテーションすると、FastMCP は以下の処理を行います。

FastMCPは、バイト列パラメータの Base64エンコードされた文字列を自動的にデコードしません。Base64エンコードされたデータを受け入れる必要がある場合は、以下に示すように手動でデコードを処理する必要があります。

・Base64-encoded strings

from typing import Annotated
from pydantic import Field

@mcp.tool()
def process_image_data(
    image_data: Annotated[str, Field(description="Base64-encoded image data")]
):
    """Base64 エンコードされた文字列から画像を処理します。

    クライアントは Base64 エンコードされたデータを文字列として提供することが想定されています。
手動でデコードする必要があります。
    """
    # 手動のbase64デコード
    import base64
    binary_data = base64.b64decode(image_data)
    # Pバイナリデータを処理...

クライアントから base64でエンコードされたバイナリデータを受信することが予想される場合は、この方法をお勧めします。

10-7. パス

pathlibモジュールのPath型は、ファイルシステムのパスに使用できます。

from pathlib import Path

@mcp.tool()
def process_file(path: Path) -> str:
    """指定されたパスにあるファイルを処理"""
    assert isinstance(path, Path)  # パスは適切に変換
    return f"Processing file at {path}"

クライアントが文字列パスを送信すると、FastMCPはそれを自動的にPathオブジェクトに変換します。

10-8. UUID

uuidモジュールのUUID型は、一意の識別子として使用できます。

import uuid

@mcp.tool()
def process_item(
    item_id: uuid.UUID  # 文字列UUIDまたはUUIDオブジェクト
) -> str:
    """指定された UUID を持つアイテムを処理"""
    assert isinstance(item_id, uuid.UUID)  # 適切にUUIDに変換
    return f"Processing item {item_id}"

クライアントが文字列 UUID (例: 「123e4567-e89b-12d3-a456-426614174000」) を送信すると、FastMCPはそれを自動的にUUIDオブジェクトに変換します。

10-9. Pydanticモデル

ネストされたフィールドと検証を含む複雑な構造化データには、Pydanticモデルを使用します。

from pydantic import BaseModel, Field
from typing import Optional

class User(BaseModel):
    username: str
    email: str = Field(description="User's email address")
    age: int | None = None
    is_active: bool = True

@mcp.tool()
def create_user(user: User):
    """Create a new user in the system."""
    # The input is automatically validated against the User model
    # Even if provided as a JSON string or dict
    # Implementation...

Pydanticモデルを使用すると、以下のメリットが得られます。

クライアントは、Pydanticモデルパラメータのデータを以下のいずれかの形式で提供できます。

10-10. Pydanticフィールド

FastMCPは、PydanticのFieldクラスを通じて堅牢なパラメータ検証をサポートしています。これは、入力値が型だけでなく特定の要件を満たしていることを確認するのに特に役立ちます。

Fieldは Pydantic モデル外でもメタデータや検証制約を提供するために使用できます。推奨される方法は、Annotated と Field を使用することです。

from typing import Annotated
from pydantic import Field

@mcp.tool()
def analyze_metrics(
    # 範囲制約のある数値
    count: Annotated[int, Field(ge=0, le=100)],         # 0 <= count <= 100
    ratio: Annotated[float, Field(gt=0, lt=1.0)],       # 0 < ratio < 1.0
    
    # パターンと長さの制約を持つ文字列
    user_id: Annotated[str, Field(
        pattern=r"^[A-Z]{2}\d{4}$",                     # 正規表現パターンに一致する必要がある
        description="User ID in format XX0000"
    )],
    
    # 長さ制約のある文字列
    comment: Annotated[str, Field(min_length=3, max_length=500)] = "",
    
    # 数値制約
    factor: Annotated[int, Field(multiple_of=5)] = 10,  # Must be multiple of 5
):
    """検証済みのパラメータを使用してメトリックを分析"""
    # 実装...

Fieldをデフォルト値として使用することもできますが、Annotated付きのアプローチが推奨されます。

@mcp.tool()
def validate_data(
    # 値の制約
    age: int = Field(ge=0, lt=120),                     # 0 <= age < 120
    
    # 文字列制約
    email: str = Field(pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$"),  # Email pattern
    
    # コレクション制約
    tags: list[str] = Field(min_length=1, max_length=10)  # 1-10 tags
):
    """フィールド検証を使用してデータを処理"""
    # 実装...

一般的な検証オプションは次のとおりです。

クライアントが無効なデータを送信すると、FastMCPはパラメータが検証に失敗した理由を説明する検証エラーを返します。

11. サーバの動作

11-1. 重複ツール

同じ名前のツールを複数登録しようとした場合のFastMCPサーバの動作を制御できます。これは、FastMCPインスタンス作成時にon_duplicate_tools引数を使用して設定します。

from fastmcp import FastMCP

mcp = FastMCP(
    name="StrictServer",
    # 重複したツール名の動作を構成
    on_duplicate_tools="error"
)

@mcp.tool()
def my_tool(): return "Version 1"

# my_tool がすでに存在し、on_duplicate_tools が 
# "error" に設定されているため、ValueError が発生
# @mcp.tool()
# def my_tool(): return "Version 2"

次回