Fastmcp

🚀 FastMCP供應鏈優化器

本項目基於Gemini AI，對FastMCP（模型上下文協議）進行了自定義實現，可用於即時供應鏈優化。該項目受Anthropic內部FastMCP系統的啟發，展示了低延遲的多工具編排能力。

🚀 快速開始

1. 安裝依賴

pip install -r requirements.txt

2. 運行應用程序

python3 flask_app.py

3. 打開瀏覽器

訪問 http://localhost:5000

4. 替代方案：使用本地大語言模型

若出於數據隱私和內部工具使用的考慮，你可以使用 local-llm-api 將Gemini API替換為你自己的本地大語言模型：

# 克隆並設置本地大語言模型API
git clone https://github.com/ANSH-RIYAL/local-llm-api.git
cd local-llm-api
./run_server.sh

# 修改fastmcp_server.py以使用本地API而非Gemini
# 將GEMINI_API_KEY替換為CUSTOM_API_URL = "http://localhost:8050"

✨ 主要特性

• 自定義FastMCP實現：在每個大語言模型處理步驟中進行多工具調用（非順序執行）
• 即時事件處理：處理供應鏈事件流，並通過即時AI響應
• 智能推薦：由AI驅動的庫存優化，提供可操作的見解
• 即時Web界面：通過美觀的用戶界面進行即時監控和控制
• 模塊化工具架構：易於擴展和修改，以適應不同的用例

🔧 關於FastMCP與MCP

FastMCP並非開源項目，它是Anthropic的內部實現。本項目是對FastMCP關鍵創新點的最小模擬：

核心區別：並行工具調用

• 標準MCP：按順序交替進行1次大語言模型調用 → 1次工具調用 → 1次大語言模型調用
• FastMCP：在大語言模型處理的每個步驟中調用多個工具
• 本實現：針對每個事件模擬FastMCP的方法，執行多個工具

由於FastMCP未開源，我構建了一個低延遲多工具編排棧的最小模擬，展示了大語言模型代理如何通過路由工具對即時供應鏈更新做出可操作的建議。

💻 使用示例

基礎用法

本項目的使用流程如下：

1. 啟動FastMCP服務器：點擊“Start FastMCP Server”初始化AI代理
2. 啟動事件流：點擊“Start Event Stream”開始處理供應鏈事件
3. 監控結果：即時查看終端輸出和行動建議
4. 完成後停止：使用停止按鈕正常關閉程序

高級用法

處理事件類型

• DEMAND_SPIKE：產品需求突然增加
• DELAY：供應商交貨延遲
• COST_INCREASE：供應商價格變動

AI採取的行動

• 庫存轉移：在倉庫之間轉移庫存
• 重新訂購建議：建議新的訂單和數量
• 缺貨預測：預測產品何時會缺貨
• 成本優化：分析供應商替代方案

📚 詳細文檔

🛠️ 已實現的工具

核心供應鏈工具

1. get_inventory_status

• 用途：檢查所有倉庫的當前庫存水平
• 參數：product_id（可選）
• 返回值：產品或所有產品的完整庫存數據
• 示例：{"product_id": "P001"} → 返回倉庫A、B、C的庫存水平

2. update_inventory

• 用途：修改倉庫庫存水平（增加/減少）
• 參數：product_id、warehouse、quantity
• 返回值：成功狀態和庫存變更詳情
• 示例：{"product_id": "P001", "warehouse": "warehouse_A", "quantity": -10}

3. calculate_transfer

• 用途：在倉庫之間轉移庫存
• 參數：product_id、from_warehouse、to_warehouse、quantity
• 返回值：轉移執行詳情和新的庫存水平
• 示例：{"product_id": "P001", "from_warehouse": "warehouse_B", "to_warehouse": "warehouse_A", "quantity": 20}

4. predict_stockout

• 用途：預測產品何時會缺貨
• 參數：product_id、warehouse
• 返回值：風險級別和預測的缺貨時間線
• 示例：{"product_id": "P001", "warehouse": "warehouse_A"} → "高風險，1 - 2天"

5. recommend_reorder

• 用途：建議重新訂購的數量和供應商
• 參數：product_id、quantity
• 返回值：包含成本計算的訂單詳情
• 示例：{"product_id": "P001", "quantity": 50} → "訂單：從供應商X訂購50件，單價5.50美元"

如何修改工具

添加新工具

1. 在 supply_chain_tools.py 中添加函數：

def new_tool_function(self, param1: str, param2: int) -> Dict[str, Any]:
    """描述此工具的作用"""
    # 實現邏輯
    return {"success": True, "result": "工具輸出"}

2. 在 fastmcp_server.py 中註冊工具：

Tool(
    name="new_tool_function",
    description="描述此工具的作用",
    inputSchema={
        "type": "object",
        "properties": {
            "param1": {"type": "string", "description": "參數1"},
            "param2": {"type": "integer", "description": "參數2"}
        },
        "required": ["param1", "param2"]
    }
)

3. 在 handle_call_tool 中添加處理程序：

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

🏗️ 架構

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   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         # Web界面
├── supply_chain_tools.py  # 核心業務邏輯
├── fastmcp_server.py      # 自定義FastMCP實現
├── flask_app.py          # Web服務器和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驅動的決策
• 即時更新：Web界面即時更新
• 簡單設置：依賴項和配置最少
• 可擴展：易於添加新工具和事件類型
• 隱私選項：可以使用本地大語言模型替代雲API

🎯 用例

• 供應鏈優化：即時庫存管理
• 需求預測：由AI驅動的庫存預測
• 成本優化：供應商和定價分析
• 風險管理：預防和緩解缺貨情況

🔄 場景修改

1. 即時供應鏈優化器（流式輸入 + 即時代理校正）

當前實現情況：✅ 部分實現

• ✅ 流式CSV事件
• ✅ 即時AI響應
• ✅ 基本庫存工具
• ❌ 快速相關性計算器
• ❌ 預測工具（ARIMA/指數平滑）
• ❌ 即時代理校正

近期可添加的功能：

# 添加到supply_chain_tools.py
def calculate_correlation(self, product1: str, product2: str) -> Dict[str, Any]:
    """計算產品之間的需求相關性"""
    # 使用pandas相關性實現

def forecast_demand(self, product_id: str, periods: int) -> Dict[str, Any]:
    """使用簡單指數平滑法預測需求"""
    # 使用statsmodels實現

def recommend_reroute(self, from_supplier: str, to_supplier: str) -> Dict[str, Any]:
    """根據延遲/成本建議供應重新路由"""
    # 實現成本分析

示例對話：

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. 交互式調查分析器（多代理 & 多工具）

需要修改的內容：

# survey_tools.py中的新工具
def extract_themes(self, responses: List[str]) -> Dict[str, Any]:
    """從調查響應中提取常見主題"""

def compute_frequencies(self, data: pd.DataFrame) -> Dict[str, Any]: 
    """計算響應頻率和置信區間"""

def generate_summary_report(self, insights: Dict) -> Dict[str, Any]:
    """生成面向客戶的摘要報告"""

示例對話：

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. 臨床分診助手（低延遲循環的工具選擇）

需要修改的內容：

# clinical_tools.py中的新工具
def check_symptoms(self, symptoms: List[str]) -> Dict[str, Any]:
    """根據醫學數據庫檢查症狀"""

def classify_risk(self, vitals: Dict) -> Dict[str, Any]:
    """對患者風險級別進行分類"""

def score_triage_priority(self, risk: str, symptoms: List) -> Dict[str, Any]:
    """對分診優先級進行評分"""

def generate_doctor_note(self, patient_data: Dict) -> Dict[str, Any]:
    """生成醫生筆記"""

示例對話：

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. 電子商務定價代理（快速反饋循環）

需要修改的內容：

# pricing_tools.py中的新工具
def calculate_optimal_price(self, cost: float, margin: float, demand_factor: float) -> Dict[str, Any]:
    """使用公式計算最優價格"""

def find_competitor_match(self, product_id: str) -> Dict[str, Any]:
    """查找最近的競爭對手產品"""

def generate_markdown_explanation(self, price_change: Dict) -> Dict[str, Any]:
    """為價格變化生成Markdown解釋"""

示例對話：

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...'"

🔧 開發

添加新工具

1. 在 supply_chain_tools.py 中添加函數
2. 在 fastmcp_server.py 中註冊工具
3. 根據需要更新系統提示

添加新事件類型

1. 在 data/events.csv 中添加事件
2. 在 fastmcp_server.py 中更新事件處理邏輯
3. 使用Web界面進行測試

切換到本地大語言模型

1. 設置 local-llm-api
2. 修改 fastmcp_server.py 以使用本地API端點
3. 更新提示以適應本地模型

📄 許可證

本項目文檔未提及許可證相關信息。

📝 注意事項

• 本項目是使用模擬數據的演示項目
• 服務器停止時，庫存更改會保存回CSV文件
• 使用Gemini API免費層（適用速率限制）
• 設計目的是為了簡單和教育
• FastMCP並非開源項目，本項目是自定義實現
• 可以使用本地大語言模型擴展以保護數據隱私

🤝 貢獻

歡迎通過以下方式擴展本項目：

• 更復雜的AI模型
• 與真實數據庫集成
• 更多供應鏈工具
• 增強Web界面功能
• 並行工具執行優化
• 即時數據流

🔗 相關項目

• local-llm-api：用於數據隱私的本地大語言模型API
• MCP-RAG：參考MCP實現

準備好使用AI優化你的供應鏈了嗎？ 啟動服務器，見證奇蹟發生！🚀

本項目展示瞭如何構建一個類似FastMCP的系統，用於即時多工具AI編排。