Kroger MCP

🚀 Kroger MCP 服務器

本項目實現了一個封裝了克羅格（Kroger）API 的模型上下文協議（MCP）服務器。它允許像 Anthropic 的 Claude 這樣的大語言模型（LLM）與克羅格的雜貨服務進行交互，實現產品搜索、店鋪查找和購物車管理等功能。

✨ 主要特性

• OAuth2 認證：處理通用數據的客戶端憑證以及購物車操作的基於用戶的授權。
• 產品搜索：按關鍵詞在特定店鋪位置搜索產品。
• 產品詳情：檢索產品的詳細信息，包括價格、可用性和配送選項。
• 店鋪位置：按郵政編碼查找克羅格店鋪位置。
• 購物車管理：將商品添加到用戶的克羅格購物車（需要用戶授權）。

📦 安裝指南

1. 配置

在運行服務器之前，你需要配置克羅格 API 憑證和 OAuth2 設置。

1.1 API 憑證（客戶端 ID 和密鑰）

1. 獲取憑證：在 克羅格開發者門戶 上註冊一個應用程序，以獲取 客戶端 ID 和 客戶端密鑰。
2. 設置憑證：你可以通過以下兩種方式之一設置這些憑證：
   • config.py（推薦用於本地簡單使用）： 打開 config.py 文件，將 KROGER_CLIENT_ID 和 KROGER_CLIENT_SECRET 的佔位符值替換為你實際的憑證。

# config.py
KROGER_CLIENT_ID = "YOUR_ACTUAL_CLIENT_ID"
KROGER_CLIENT_SECRET = "YOUR_ACTUAL_CLIENT_SECRET"

- **環境變量**：如果你更喜歡通過環境變量管理機密信息，可以修改 `auth.py` 腳本，使其也能讀取 `os.environ.get("KROGER_CLIENT_ID")` 和 `os.environ.get("KROGER_CLIENT_SECRET")`。（注意：`tools.py` 和 `server.py` 當前的實現直接使用 `config.py` 中的值）。

1.2 用戶授權（用於購物車操作）

要使用修改用戶購物車的工具（例如 add_to_cart），用戶必須授權該應用程序。此服務器使用 OAuth2 授權碼授予流程。

1. 重定向 URI：確保 config.py 中的 KROGER_REDIRECT_URI 與你在克羅格應用程序中註冊的重定向 URI 相匹配。對於本地測試，http://localhost:8080/callback 是常見的默認值，但你需要一種方法來捕獲此重定向中的代碼。

# config.py
KROGER_REDIRECT_URI = "http://localhost:8080/callback" # 或你配置的 URI

2. 獲取授權碼和刷新令牌：
   • 直接運行 auth.py 腳本（python auth.py）。
   • 它將打印出一個“授權 URL”。將此 URL 複製並粘貼到你的網頁瀏覽器中。
   • 使用你的克羅格賬戶登錄並授予訪問權限。
   • 你將被重定向到你的 KROGER_REDIRECT_URI。瀏覽器地址欄中的 URL 現在將包含一個授權 代碼（例如，http://localhost:8080/callback?code=YOUR_AUTH_CODE&...）。
   • 複製此 代碼。
   • 在提示時將 代碼 粘貼回 auth.py 腳本中。
   • 腳本將用該代碼交換一個訪問令牌和一個 刷新令牌。
   • 至關重要的是，安全地存儲顯示的 刷新令牌。
3. 配置刷新令牌：
   • 為了在服務器重啟時無需每次都重新認證即可啟用購物車操作，你應該將獲取的 刷新令牌 設置在 config.py 中，或者作為 AuthManager 可以加載的環境變量。
   • 修改 auth.py 中的 AuthManager.__init__ 以從 config.py 或環境中加載此 KROGER_USER_REFRESH_TOKEN：

# 在 auth.py -> AuthManager.__init__ 中
# self.user_refresh_token = os.environ.get("KROGER_USER_REFRESH_TOKEN") 
# 或者
# from config import KROGER_USER_REFRESH_TOKEN # 添加到 config.py
# self.user_refresh_token = KROGER_USER_REFRESH_TOKEN 

然後在 config.py 中添加 KROGER_USER_REFRESH_TOKEN = "YOUR_SAVED_REFRESH_TOKEN"。 - 當調用 get_user_token() 時，如果訪問令牌過期或缺失，它將嘗試使用此刷新令牌。

2. 運行服務器

1. 安裝依賴項：如果你還沒有安裝所需的 Python 庫，請安裝：

pip install requests mcp

（注意：假設 mcp 庫的名稱如此；如果不同，請進行調整，例如 modelcontextprotocol）。 2. 啟動服務器： 在終端中運行 server.py 腳本：

python server.py

3. 服務器操作：
   • 服務器使用 標準輸入輸出（STDIO） 與 MCP 客戶端進行通信。它不會打開任何網絡端口。
   • 啟動後，它將打印初始化消息，包括已註冊工具的列表。
   • 然後它將監聽來自 MCP 客戶端的 JSON-RPC 請求。
4. 停止服務器： 在運行服務器的終端中按 Ctrl+C。

💻 使用示例

3. MCP 客戶端集成

3.1 Claude 桌面版

• 打開 Claude 桌面版的 設置。
• 導航到 集成（或類似的 MCP 服務器部分）。
• 點擊 添加 MCP 服務器（或等效操作）。
• 提供運行服務器的命令。通常需要指定 Python 解釋器和 server.py 的路徑。例如：
  • 如果 Python 在你的系統路徑中：python /path/to/your/project/server.py
  • 否則：/path/to/your/python /path/to/your/project/server.py
• 添加後，Claude 將能夠查看和調用克羅格工具（例如 find_stores、search_products）。

3.2 編程式使用（示例）

開發者還可以使用 MCP 客戶端庫以編程方式與服務器進行交互。

# 這是一個基於 MCP 規範的概念示例。
# 實際的庫可能會有所不同。
from modelcontext import Client, StdioClientTransport # 假設庫結構

async def main():
    client = Client(name="example-kroger-client", version="1.0", capabilities={})
    
    # 如果 python/server.py 不在系統路徑中或需要完整路徑，請調整命令
    python_executable = "python" # 或 Python 解釋器的完整路徑
    server_script_path = "server.py" # 或 server.py 的完整路徑
    
    transport = StdioClientTransport(command=[python_executable, server_script_path])
    
    await client.connect(transport)
    
    # 示例：查找店鋪
    try:
        store_results = await client.call_tool(
            "find_stores", 
            {"zip_code": "45202", "limit": 1}
        )
        print("店鋪搜索結果:", store_results)

        if store_results and not store_results.get("error") and len(store_results) > 0:
            location_id = store_results[0].get("locationId")
            if location_id:
                # 示例：搜索產品
                product_results = await client.call_tool(
                    "search_products",
                    {"query": "milk", "location_id": "01400123", "limit": 2}
                )
                print("產品搜索結果:", product_results)
                
                # 示例：添加到購物車（需要在服務器中設置用戶認證令牌）
                # 確保 product_results[0] 存在且有 'productId'
                if product_results and not product_results.get("error") and len(product_results) > 0:
                    product_id = product_results[0].get("productId")
                    cart_result = await client.call_tool(
                        "add_to_cart",
                        {"product_id": product_id, "quantity": 1, "location_id": location_id}
                    )
                    print("添加到購物車結果:", cart_result)

    except Exception as e:
        print(f"發生錯誤: {e}")
    finally:
        await client.disconnect()

if __name__ == "__main__":
    # 如果你的客戶端庫使用 asyncio
    # import asyncio
    # asyncio.run(main())
    print("如果你的 MCP 客戶端庫需要，請使用 asyncio 事件循環運行異步 main() 函數。")

4. 與大語言模型的示例對話

用戶：“我需要兩加侖有機全脂牛奶和一打雞蛋，從 90210 附近的克羅格店購買。”

大語言模型（助手）內部步驟：

1. （可選：如果未提供或不明確，大語言模型詢問用戶的郵政編碼）
2. 大語言模型調用 find_stores：{"zip_code": "90210", "limit": 1}
   • 服務器返回店鋪詳情，例如 [{ "locationId": "01400123", "name": "比佛利山莊克羅格店", ... }]
3. 大語言模型調用 search_products（搜索牛奶）：{"query": "organic whole milk", "location_id": "01400123", "limit": 5}
   • 服務器返回牛奶產品列表。大語言模型選擇一個，例如 {"productId": "0001111060404", "description": "Simple Truth 有機牛奶...", ...}
4. 大語言模型調用 search_products（搜索雞蛋）：{"query": "dozen eggs", "location_id": "01400123", "limit": 3}
   • 服務器返回雞蛋產品列表。大語言模型選擇一個。
5. （如果尚未完成，用戶必須完成購物車的授權）
6. 大語言模型調用 add_to_cart（添加牛奶）：{"product_id": "0001111060404", "quantity": 2, "location_id": "01400123"}
   • 服務器確認添加。
7. 大語言模型調用 add_to_cart（添加雞蛋）：{"product_id": "...", "quantity": 1, "location_id": "01400123"}
   • 服務器確認添加。

大語言模型（助手）回覆用戶：“好的，我找到了比佛利山莊克羅格店。我已經將兩加侖 Simple Truth 有機全脂牛奶和一打雞蛋添加到你的購物車中。還有其他需求嗎？”

📚 詳細文檔

5. 錯誤場景

• 缺少用戶授權：如果你在用戶未授權應用程序的情況下嘗試使用 add_to_cart，該工具將返回錯誤：

{
  "error": "需要用戶認證。",
  "message": "未找到用戶訪問令牌。用戶需要授權該應用程序...",
  "action_needed": "用戶必須完成 OAuth2 授權流程。" 
}

大語言模型應引導用戶執行授權步驟（見第 1.2 節）。錯誤消息中可能會包含授權 URL。

• 無效/過期的令牌：如果訪問令牌過期，AuthManager 將嘗試刷新它。如果刷新令牌也無效（例如，在 add_to_cart 出現 401 錯誤後針對用戶令牌），則需要重新授權。
• API 速率限制：克羅格的 API 有速率限制（例如，見 developer.kroger.com/support/rate-limits/）。如果服務器達到這些限制，API 調用將失敗。服務器將返回來自克羅格的錯誤，通常狀態碼為 HTTP 429。大語言模型應告知用戶稍後再試。
• 其他 API 錯誤：如果克羅格的 API 返回其他錯誤（例如，無效的產品 ID、根據 locationId 未找到店鋪），工具將返回一個包含 error、details、status_code（來自克羅格的 HTTP 狀態）以及可能的 raw_response 或 kroger_error 字段的 JSON 字典。

6. 可用工具

服務器向大語言模型公開以下工具：

• find_stores(zip_code: str, radius_miles: int = 10, limit: int = 5) -> list | dict
  • 描述：按郵政編碼查找克羅格店鋪位置（返回帶有 ID 的最近店鋪）。
• search_products(query: str, location_id: str, limit: int = 10) -> list | dict
  • 描述：在給定店鋪按關鍵詞搜索克羅格產品。
• get_product(product_id: str, location_id: str) -> dict
  • 描述：按 ID 獲取產品的詳細信息（價格、尺寸、庫存、配送選項）。
• add_to_cart(product_id: str, quantity: int, location_id: str) -> dict
  • 描述：將產品添加到用戶的克羅格購物車（需要用戶認證）。

（以上描述基於 tools.py 中的 @tool 裝飾器）。