Fastmcp

🚀 FastMCP サプライチェーンオプティマイザ

Gemini AIを使用したリアルタイムサプライチェーン最適化のための FastMCP (モデルコンテキストプロトコル) の カスタム実装 です。このプロジェクトは、Anthropic社の内部FastMCPシステムにインスパイアされた低レイテンシのマルチツールオーケストレーションを実証しています。

🚀 クイックスタート

1. 依存関係のインストール

pip install -r requirements.txt

2. アプリケーションの実行

python3 flask_app.py

3. ブラウザを開く

http://localhost:5000 にアクセスします。

4. 代替案: ローカルLLMの使用

データのプライバシーや内部ツールの使用のために、local-llm-api を使用してGemini APIを独自のローカルLLMに置き換えることができます。

# ローカルLLM APIをクローンしてセットアップ
git clone https://github.com/ANSH-RIYAL/local-llm-api.git
cd local-llm-api
./run_server.sh

# fastmcp_server.pyを修正してGeminiの代わりにローカルAPIを使用する
# GEMINI_API_KEYをCUSTOM_API_URL = "http://localhost:8050"に置き換える

✨ 主な機能

• カスタムFastMCP実装：各LLM処理ステップでのマルチツール呼び出し（逐次ではない）
• リアルタイムイベント処理：サプライチェーンイベントのストリームとライブAI応答
• インテリジェントな推奨：AIによる在庫最適化と実行可能な洞察
• ライブウェブインターフェイス：美しいUIでのリアルタイム監視と制御
• モジュール型ツールアーキテクチャ：異なるユースケースに容易に拡張および修正可能

📦 インストール

依存関係のインストールとアプリケーションの実行については、「クイックスタート」を参照してください。

💻 使用例

基本的な使用法

1. FastMCPサーバーを起動する：「Start FastMCP Server」をクリックしてAIエージェントを初期化します。
2. イベントストリームを開始する：「Start Event Stream」をクリックしてサプライチェーンイベントの処理を開始します。
3. 結果を監視する：ターミナル出力とアクションの推奨をリアルタイムで確認します。
4. 終了時に停止する：停止ボタンを使用して正常にシャットダウンします。

高度な使用法

ツールの追加

1. supply_chain_tools.pyに関数を追加する：

def new_tool_function(self, param1: str, param2: int) -> Dict[str, Any]:
    """Description of what this tool does"""
    # Implementation logic
    return {"success": True, "result": "tool output"}

2. fastmcp_server.pyでツールを登録する：

Tool(
    name="new_tool_function",
    description="Description of what this tool does",
    inputSchema={
        "type": "object",
        "properties": {
            "param1": {"type": "string", "description": "Parameter 1"},
            "param2": {"type": "integer", "description": "Parameter 2"}
        },
        "required": ["param1", "param2"]
    }
)

3. handle_call_toolにハンドラを追加する：

elif name == "new_tool_function":
    result = self.tools.new_tool_function(
        arguments["param1"],
        arguments["param2"]
    )

📚 ドキュメント

処理されるイベントタイプ

• DEMAND_SPIKE：製品の需要が急増する
• DELAY：サプライヤーの納品が遅れる
• COST_INCREASE：サプライヤーからの価格変更

AIアクション

• 在庫移動：倉庫間で在庫を移動する
• 再注文推奨：数量を指定した新しい注文を提案する
• 欠品予測：製品が在庫切れになる時期を予測する
• コスト最適化：サプライヤーの代替案を分析する

アーキテクチャ

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Flask Web     │    │   Custom        │    │   Gemini AI     │
│   Interface     │◄──►│   FastMCP       │◄──►│   (or Local     │
│                 │    │   Server        │    │    LLM API)     │
└─────────────────┘    └─────────────────┘    └─────────────────┘
         │                       │
         │                       │
         ▼                       ▼
┌─────────────────┐    ┌─────────────────┐
│   Event Stream  │    │   Supply Chain  │
│   (CSV Data)    │    │   Tools         │
└─────────────────┘    └─────────────────┘

プロジェクト構造

FastMCP/
├── data/
│   ├── inventory.csv      # 製品在庫データ
│   └── events.csv         # サプライチェーンイベントストリーム
├── templates/
│   └── index.html         # ウェブインターフェイス
├── supply_chain_tools.py  # コアビジネスロジック
├── fastmcp_server.py      # カスタムFastMCP実装
├── flask_app.py          # ウェブサーバーとAPI
├── test_demo.py          # デモスクリプト
├── requirements.txt      # Python依存関係
└── README.md            # このファイル

例のワークフロー

1. イベント：DEMAND_SPIKE for P001 - 40 units
2. 分析：AIがすべての倉庫の現在の在庫を確認する
3. 予測：潜在的な欠品リスクを特定する
4. アクション：倉庫BからAへの移動を推奨する
5. 実行：在庫を更新し、アクションを記録する

サンプル会話フロー

Event Stream → MCP Client: "DEMAND_SPIKE: P001, 40 units"
MCP Client → get_inventory_status: {"product_id": "P001"}
MCP Client → predict_stockout: {"product_id": "P001", "warehouse": "warehouse_A"}
MCP Client → calculate_transfer: {"product_id": "P001", "from_warehouse": "warehouse_B", "to_warehouse": "warehouse_A", "quantity": 20}
MCP Client → recommend_reorder: {"product_id": "P001", "quantity": 50}
MCP Client → User: "Transfer 20 units from B to A, reorder 50 units from Supplier X"

監視

• ターミナル出力：リアルタイムのサーバーログと処理ステータス
• アクションログ：すべてのAI推奨と実行されたアクション
• ステータスインジケーター：サーバーとイベントストリームのステータス
• イベント進捗：現在処理中のイベント

使用例

• サプライチェーン最適化：リアルタイムの在庫管理
• 需要予測：AIによる在庫予測
• コスト最適化：サプライヤーと価格の分析
• リスク管理：欠品の防止と軽減

シナリオの変更

1. リアルタイムサプライチェーンオプティマイザ（ストリーミング入力 + ライブエージェント修正）

現在の実装：✅ 部分的に実装済み

• ✅ ストリーミングCSVイベント
• ✅ リアルタイムAI応答
• ✅ 基本的な在庫ツール
• ❌ 高速相関計算機
• ❌ 予測ツール（ARIMA/指数平滑法）
• ❌ ライブエージェント修正

すぐに追加できるもの：

# Add to supply_chain_tools.py
def calculate_correlation(self, product1: str, product2: str) -> Dict[str, Any]:
    """Calculate demand correlation between products"""
    # Implementation using pandas correlation

def forecast_demand(self, product_id: str, periods: int) -> Dict[str, Any]:
    """Forecast demand using simple exponential smoothing"""
    # Implementation using statsmodels

def recommend_reroute(self, from_supplier: str, to_supplier: str) -> Dict[str, Any]:
    """Recommend supply rerouting based on delays/costs"""
    # Implementation with cost analysis

例の会話：

Event: "SUPPLIER_DELAY: Supplier X, 3 days"
MCP Client: "Analyzing impact on P001, P002, P003..."
Tools Called: [get_inventory_status, calculate_correlation, forecast_demand, recommend_reroute]
Response: "Reroute P001 from Supplier X to Supplier Y. P002 and P003 show 0.8 correlation - adjust P002 orders accordingly."

2. インタラクティブサーベイアナライザー（マルチエージェント & マルチツール）

必要な変更：

# New tools in survey_tools.py
def extract_themes(self, responses: List[str]) -> Dict[str, Any]:
    """Extract common themes from survey responses"""

def compute_frequencies(self, data: pd.DataFrame) -> Dict[str, Any]:
    """Compute response frequencies and confidence intervals"""

def generate_summary_report(self, insights: Dict) -> Dict[str, Any]:
    """Generate client-facing summary reports"""

例の会話：

User: "Analyze 500 survey responses about Product X"
MCP Client: "Processing responses with multiple agents..."
Tools Called: [extract_themes, compute_frequencies, generate_summary_report]
Response: "Top themes: UI/UX (45%), Performance (32%), Price (23%). 78% satisfaction rate (±3% CI). Report generated."

3. 臨床トリアージアシスタント（低レイテンシループでのツール選択）

必要な変更：

# New tools in clinical_tools.py
def check_symptoms(self, symptoms: List[str]) -> Dict[str, Any]:
    """Check symptoms against medical database"""

def classify_risk(self, vitals: Dict) -> Dict[str, Any]:
    """Classify patient risk level"""

def score_triage_priority(self, risk: str, symptoms: List) -> Dict[str, Any]:
    """Score triage priority"""

def generate_doctor_note(self, patient_data: Dict) -> Dict[str, Any]:
    """Generate doctor notes"""

例の会話：

Patient Data: {"symptoms": ["chest pain", "shortness of breath"], "vitals": {"bp": "140/90"}}
MCP Client: "Analyzing patient data..."
Tools Called: [check_symptoms, classify_risk, score_triage_priority, generate_doctor_note]
Response: "HIGH RISK - Cardiac symptoms detected. Immediate triage required. Doctor note: 'Patient presents with chest pain and elevated BP...'"

4. 電子商取引価格設定エージェント（高速フィードバックループ）

必要な変更：

# New tools in pricing_tools.py
def calculate_optimal_price(self, cost: float, margin: float, demand_factor: float) -> Dict[str, Any]:
    """Calculate optimal price using formula"""

def find_competitor_match(self, product_id: str) -> Dict[str, Any]:
    """Find nearest competitor product"""

def generate_markdown_explanation(self, price_change: Dict) -> Dict[str, Any]:
    """Generate markdown explanation for price changes"""

例の会話：

Event: "COMPETITOR_PRICE_CHANGE: Product X, $25.99 → $22.99"
MCP Client: "Analyzing competitive landscape..."
Tools Called: [find_competitor_match, calculate_optimal_price, generate_markdown_explanation]
Response: "Competitor reduced price by 12%. Recommended action: Reduce price to $23.99. Explanation: 'We've adjusted our pricing to remain competitive while maintaining healthy margins...'"

🔧 技術詳細

FastMCPとMCPの比較

FastMCPはオープンソースではありません - これはAnthropic社の内部実装です。このプロジェクトはFastMCPの主要な革新点を 最小限にシミュレート したものです。

核心的な違い：並列ツール呼び出し

• 標準MCP：1回のLLM呼び出し → 1回のツール呼び出し → 1回のLLM呼び出しを逐次的に繰り返す
• FastMCP：LLM処理の各ステップで複数のツールを呼び出す
• この実装：イベントごとに複数のツールを実行することでFastMCPのアプローチをシミュレートする

開発

新しいツールの追加

1. supply_chain_tools.pyに関数を追加する
2. fastmcp_server.pyでツールを登録する
3. 必要に応じてシステムプロンプトを更新する

新しいイベントタイプの追加

1. data/events.csvにイベントを追加する
2. fastmcp_server.pyのイベント処理ロジックを更新する
3. ウェブインターフェイスでテストする

ローカルLLMへの切り替え

1. local-llm-apiをセットアップする
2. fastmcp_server.pyを修正してローカルAPIエンドポイントを使用する
3. ローカルモデルとの互換性のためにプロンプトを更新する

📄 ライセンス

ドキュメントにはライセンス情報が記載されていません。

📝 注意事項

• これはシミュレートデータを使用した デモ です。
• サーバー停止時に在庫の変更がCSVに保存されます。
• Gemini APIの無料枠を使用しています（レート制限が適用されます）。
• シンプルさと教育目的で設計されています。
• FastMCPはオープンソースではありません - これはカスタム実装です。
• データのプライバシーのためにローカルLLMで拡張することができます。

🤝 コントリビューション

• より高度なAIモデル
• 実際のデータベース統合
• 追加のサプライチェーンツール
• 強化されたウェブインターフェイス機能
• 並列ツール実行の最適化
• リアルタイムデータストリーミング

で拡張することができます。

🔗 関連プロジェクト

• local-llm-api：データのプライバシーのためのローカルLLM API
• MCP-RAG：参照MCP実装

AIでサプライチェーンを最適化したいですか？ サーバーを起動して魔法を体感してください！ 🚀

このプロジェクトは、リアルタイムのマルチツールAIオーケストレーションのためのカスタムFastMCPライクなシステムの構築方法を実証しています。