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 装饰器）。