inventory-mcp

FastMCP を使用して Python で開発された、在庫照会のためのデモ MCP サーバーです。このプロジェクトは、トランスポート、MCP インターフェース、ビジネスルール、検証、データを分離し、Model Context Protocol (MCP) の主要概念の学習を支援します。

現在のスコープは意図的に読み取り専用です。サーバーは在庫の製品と数量を照会できますが、登録、変更、削除の操作はありません。

テクノロジー

• Python 3.11+

• FastMCP

• Pydantic

• pytest

• Ruff

Related MCP server: vanam-erp-mcp

アーキテクチャ

• app/server.py: FastMCP サーバーを作成し、ツールを登録して、stdio または SSE トランスポートを起動します。

• app/client.py: stdio または SSE 経由でツールを一覧表示して呼び出すデモクライアントです。

• app/tools/: MCP インターフェース。入力を検証し、サービスに委譲し、予期されるエラーを安定したレスポンスに変換します。

• app/services/: 在庫の照会と読み込みのルール。

• app/schemas/: 製品と在庫のコントラクトを定義および検証する Pydantic モデル。

• app/data/: ローカルデータソース。現在は inventory.json ファイル。

• tests/: サービス、ツール、サーバー設定の自動テスト。

Client → MCP Server → Tool → InventoryService → inventory.json

ツールはファイルに直接アクセスしません。ビジネスルールを InventoryService に委譲します。

MCP ツール

get_product

• 目的: 製品名で完全なデータを照会します。

• 入力: name (空でない string)。

• 成功時の出力: name, quantity, price を持つオブジェクト。

• 製品が存在しない場合の出力: error: "product_not_found" と説明用の message を持つオブジェクト。

• MCP 説明: Use this tool to retrieve the complete data of a product by name, including its price and stock quantity.

• 分類: 読み取り専用。

{
  "name": "Mouse",
  "quantity": 25,
  "price": 89.9
}

get_stock

• 目的: 製品名で現在の数量のみを照会します。

• 入力: name (空でない string)。

• 成功時の出力: quantity を持つオブジェクト。

• 製品が存在しない場合の出力: error: "product_not_found" と説明用の message を持つオブジェクト。

• MCP 説明: Use this tool to retrieve only the current stock quantity of a product by name.

• 分類: 読み取り専用。

{
  "quantity": 25
}

入力検証

ツールは、name が内容のある文字列であることを要求します。空の名前やスペースのみで構成された名前は、照会前に拒否されます。サービスは strip() で両端のスペースを除去し、casefold() で大文字と小文字を区別せずに名前を比較します。

Pydantic は、JSON から読み込まれたレコードと出力モデルを検証します。製品は、空でない名前、非負の整数の数量、非負の数値の価格を持つ必要があります。空の照会名の拒否は _validate_product_name() によって行われます。無効なレコードは、明示的なエラーで読み込みを中断します。

エラーハンドリング

InventoryService は、要求された製品が見つからない場合に ProductNotFoundError をスローします。ツールはこの予期されるエラーをキャッチし、予測可能なペイロードを返します:

{
  "error": "product_not_found",
  "message": "Product not found: Monitor"
}

空の名前や文字列以外の値などの入力エラーは隠蔽されません。ツール呼び出しのエラーとして報告されます。

MCP トランスポート

• stdio: 標準入力と標準出力で通信します。このプロジェクトでは、クライアントが FastMCP サーバーをサブプロセスとして起動し、呼び出しを実行して、終了時にプロセスを終了します。

• SSE: Server-Sent Events を使用した HTTP エンドポイントで通信します。サーバーとクライアントは別々のプロセスで実行されます。デフォルトでは、サーバーは http://127.0.0.1:8000/sse で待ち受けます。

実行方法

以下のコマンドは PowerShell を使用し、プロジェクトのルートで実行する必要があります。

仮想環境の作成とアクティブ化

python -m venv .venv
.\.venv\Scripts\Activate.ps1

依存関係のインストール

python -m pip install --upgrade pip
python -m pip install -e ".[dev]"

stdio 経由で実行

クライアントはデフォルトで stdio を使用し、サーバーをサブプロセスとして起動します:

.\.venv\Scripts\python.exe -m app.client

サーバーのみを直接起動するには:

.\.venv\Scripts\python.exe -m app.server --transport stdio

SSE 経由で実行

ターミナルでサーバーを起動します (sse はサーバーのデフォルトトランスポートです):

.\.venv\Scripts\python.exe -m app.server

同等の明示的なコマンドは python -m app.server --transport sse です。別のターミナルで、クライアントを接続します:

.\.venv\Scripts\python.exe -m app.client --transport sse

クライアントは --url を使用して別のエンドポイントを受け付けます。

テストの実行

.\.venv\Scripts\pytest.exe

Ruff の実行

.\.venv\Scripts\ruff.exe check .
.\.venv\Scripts\ruff.exe format --check .

ツールリスク評価

現在のツールは読み取り専用であり、データを作成、変更、削除することはできません。この決定によりリスク面は減りますが、機密性と可用性への潜在的な影響は排除されません。

大量の呼び出しは依然としてサーバーリソースを消費する可能性があります。ツールまたは返されるデータの将来の変更には、新しいリスク評価を伴う必要があります。

トラストバウンダリ

MCP クライアントから受け取った引数は、信頼できない入力として扱われます。

MCP Client
    ↓
MCP Server
    ↓
Tool
    ↓
InventoryService
    ↓
inventory.json

検証は、引数がサービス層で使用される前に行われます。サーバーは、MCP プロトコルを介して送信されたデータが、その理由だけで有効であるとは想定しません。inventory.json のレコードも外部入力として扱われ、読み込み中に Pydantic によって検証されます。

MCP ツールアノテーション

ツールは、その動作に応じて意味的に分類されます。現在の 2 つの操作は次のように宣言しています:

readOnlyHint=true
openWorldHint=false

readOnlyHint=true は、操作が状態を変更することを意図していないことを MCP クライアントに通知します。

openWorldHint=false は、外部システムやオープンソースを参照するのではなく、ツールが閉じた既知のドメイン（この場合はローカル在庫）で動作することを示します。

これらのアノテーションは、MCP クライアントのためのメタデータとヒントとして機能し、セキュリティメカニズムとしては機能しません。クライアントは、検証、認可、その他の実際の制御の代わりとしてこれらを信頼すべきではありません。

書き込みツールのリスク

次のような将来の操作は:

update_stock(name, quantity)

システムの永続状態を変更するため、リスクが大幅に高くなります。

誤った呼び出しや悪意のある呼び出しは、誤った製品を変更したり、無効な値を記録したり、不正な変更を許可したりする可能性があります。update_stock のような将来のツールには、厳格な検証、認証、認可、監査、トレーシングが必要です。破壊的な操作には、該当する場合、確認または承認も必要です。

トランスポートごとのリスク

stdio では、サーバーはクライアントのサブプロセスとしてローカルに起動されるため、ネットワーク露出が軽減されます。SSE では、サーバーとクライアントは別々のプロセスであり、通信は HTTP エンドポイントを使用します。このエンドポイントをローカルホストの外部に公開する場合は、追加のアクセス制御と可用性制御が必要になります。

テスト

現在のスイートは以下を検証します:

• InventoryService の読み込み、検索、正規化、エラー;

• ツールの戻り値と、存在しない製品の予測可能なエラーへの変換;

• 空の名前と文字列以外の値の拒否;

• Pydantic による無効な在庫レコードの拒否;

• サーバーへのツールの登録;

• SSE および stdio トランスポートの選択と設定;

• list_tools()、get_stock の呼び出し、MCP アノテーションの読み取りを含む、stdio 経由の実際の統合。

シナリオには、存在する製品と存在しない製品、両端のスペース、大文字と小文字の違い、無効な入力が含まれます。エンドツーエンドテストでは、実際の FastMCP クライアントがサーバーをサブプロセスとして起動し、readOnlyHint と openWorldHint を検証し、ローカル JSON から読み込まれた在庫を照会し、コンテキストマネージャーによって接続を終了します。

コード品質

このプロジェクトは型ヒントを使用し、MCP、サービス、スキーマ、データ間で責任を分離し、依存関係を最小限に保ちます。pytest は実装された動作をカバーし、Ruff は lint、import、Python 3.11 との互換性、フォーマットを検証します。

現在の制限事項

• データはローカルの JSON ファイルから読み込まれます。

• データベースはありません。

• AI や LLM との統合はありません。

• 書き込みツールはありません。

• 認証や認可はありません。

考えられる今後の発展

• トレーシングと構造化ロギング（プロジェクトの教育的焦点を保つため、現在のスコープ外）;

• Streamable HTTP のサポート;

• データベースへの永続化;

• 認証と認可;

• セーフガード付きの書き込みツール;

• LLM との将来の統合。