MCP Waifu Queue

MCP Waifu Queue（双子版）

该项目为对话式 AI 角色“waifu”实现了一个 MCP（模型上下文协议）服务器，利用 Google Gemini API 通过 Redis 队列进行异步处理。它利用FastMCP库简化了服务器的设置和管理。

目录

• 特征
• 建筑学
• 先决条件
• 安装
• 配置
• 运行服务
• MCP API
• 测试
• 故障排除
• 贡献
• 执照

特征

• 使用 Google Gemini API ( gemini-2.5-pro-preview-03-25 ) 生成文本。
• 使用 Redis 进行请求排队，以异步处理并发请求。
• 使用FastMCP MCP 兼容 API。
• 通过 MCP 资源跟踪作业状态。
• 通过环境变量（ .env文件）进行配置，并从~/.api-gemini加载 API 密钥。

建筑学

该项目由几个关键部分组成：

• main.py ：主入口点，初始化FastMCP应用程序并定义 MCP 工具/资源。
• respond.py ：包含使用google-generativeai库与 Gemini API 交互的核心文本生成逻辑。
• task_queue.py ：处理与 Redis 队列的交互（使用python-rq ），将生成请求排入队列。
• utils.py ：包含实用函数，特别是call_predict_response ，它由工作者执行以调用respond.py中的 Gemini 逻辑。
• worker.py ：一个 Redis 工作器（ python-rq ），用于处理队列中的作业，调用call_predict_response 。
• config.py ：使用pydantic-settings管理配置。
• models.py ：定义用于 MCP 请求和响应验证的 Pydantic 模型。

请求的流程如下：

1. 客户端向generate_text MCP 工具（在main.py中定义）发送请求。
2. 该工具将请求（提示）排入 Redis 队列（由task_queue.py处理）。
3. worker.py进程从队列中获取作业。
4. 工作者执行call_predict_response函数（来自utils.py ）。
5. call_predict_response调用predict_response函数（在respond.py中），该函数与 Gemini API 交互。
6. 生成的文本（或错误消息）由predict_response返回，并由 RQ 存储为作业结果。
7. 客户端可以使用job://{job_id} MCP 资源（在main.py中定义）检索作业状态和结果。

先决条件

• Python 3.7+
• pip或uv （Python 包安装程序）
• Redis 服务器（已安装并运行）
• Google Gemini API 密钥

您可以在 Redis 官方网站上找到在系统上安装 Redis 的说明： https://redis.io/docs/getting-started/您可以从 Google AI Studio 获取 Gemini API 密钥： https://aistudio.google.com/app/apikey

安装

1. 克隆存储库：
   git clone <YOUR_REPOSITORY_URL> cd mcp-waifu-queue
2. 创建并激活虚拟环境（使用venv或uv ）：使用venv （标准库）：
   python -m venv .venv source .venv/bin/activate # On Linux/macOS # .venv\Scripts\activate # On Windows CMD # source .venv/Scripts/activate # On Windows Git Bash/PowerShell Core
   使用uv （如果已安装）：
   # Ensure uv is installed (e.g., pip install uv) python -m uv venv .venv source .venv/bin/activate # Or equivalent activation for your shell
3. 安装依赖项（使用 venv 或uv中的pip ）：使用pip ：
   pip install -e .[test] # Installs package in editable mode with test extras
   使用uv ：
   # Ensure uv is installed inside the venv if desired, or use the venv's python # .venv/Scripts/python.exe -m pip install uv # Example for Windows .venv/Scripts/python.exe -m uv pip install -e .[test] # Example for Windows # python -m uv pip install -e .[test] # If uv is in PATH after venv activation

配置

1. **API 密钥：**在您的主目录 ( ~/.api-gemini ) 中创建一个名为.api-gemini的文件，并将您的 Google Gemini API 密钥放入其中。确保文件没有多余的空格。
   echo "YOUR_API_KEY_HERE" > ~/.api-gemini
   （将YOUR_API_KEY_HERE替换为您的实际密钥）
2. **其他设置：**将.env.example文件复制到.env ：
   cp .env.example .env
3. 修改.env文件以设置其余配置值：
   • MAX_NEW_TOKENS ：Gemini 响应的最大令牌数（默认值： 2048 ）。
   • REDIS_URL ：Redis 服务器的 URL（默认值： redis://localhost:6379 ）。
   • FLASK_ENV , FLASK_APP ：可选，如果在其他地方使用，则与 Flask 相关，而不是 MCP 服务器/工作器操作的核心。

运行服务

1. **确保 Redis 正在运行。**如果您在本地安装了 Redis，则可能需要启动 Redis 服务器进程（例如，使用redis-server命令或通过服务管理器）。
2. **启动 RQ Worker：**打开终端，激活虚拟环境（ source .venv/bin/activate或类似命令），然后运行：
   python -m mcp_waifu_queue.worker
   此命令将启动工作进程，该进程将监听.env文件中定义的 Redis 队列中的作业。请保持此终端处于运行状态。
3. **启动 MCP 服务器：**打开另一个终端，激活虚拟环境，然后使用uvicorn等工具运行 MCP 服务器（您可能需要安装它： pip install uvicorn或uv pip install uvicorn ）：
   uvicorn mcp_waifu_queue.main:app --reload --port 8000 # Example port
   将8000替换为你想要的端口。-- --reload标志对于开发很有用。或者，您可以使用start-services.sh脚本（主要为 Linux/macOS 环境设计），该脚本尝试启动 Redis（如果未运行）和后台工作程序：
   # Ensure the script is executable: chmod +x ./scripts/start-services.sh ./scripts/start-services.sh # Then start the MCP server manually as shown above.

MCP API

该服务器提供以下符合 MCP 的端点：

工具

• generate_text
  • **描述：**通过后台队列向 Gemini API 发送文本生成请求。
  • 输入： {"prompt": "Your text prompt here"} （类型： GenerateTextRequest ）
  • 输出： {"job_id": "rq:job:..."} （排队作业的唯一 ID）

资源

• job://{job_id}
  • **描述：**检索先前提交的作业的状态和结果。
  • URI参数： job_id （ generate_text工具返回的ID）。
  • 输出： {"status": "...", "result": "..."} （类型： JobStatusResponse ）
    • status ：作业的当前状态（例如，“已排队”、“已启动”、“已完成”、“已失败”）。RQ 内部使用的术语略有不同（“已启动”与“正在处理”、“已完成”与“已完成”）。资源会映射这些术语。
    • result ：如果作业状态为“已完成”，则为 Gemini 生成的文本，否则为null 。如果作业失败，则结果可能为null或包含错误信息，具体取决于 RQ 的处理方式。

测试

该项目包含测试。请确保已安装测试依赖项（ pip install -e .[test]或uv pip install -e .[test] ）。

使用pytest运行测试：

**注意：**测试可能需要模拟 Redis（ fakeredis ）并可能需要 Gemini API 调用，具体取决于它们的实现。

故障排除

• 错误： Gemini API key not found in .../.api-gemini or GEMINI_API_KEY environment variable ：请确保您已在主目录中创建~/.api-gemini文件，并将有效的 Gemini API 密钥放入其中。或者，请确保已将GEMINI_API_KEY环境变量设置为备用。
• 调用 Gemini API 时出错（例如，AuthenticationError、PermissionDenied） ：请仔细检查~/.api-gemini中的 API 密钥（或 fallback 环境变量）是否正确有效。如果适用，请确保您的 Google Cloud 项目已启用该 API。
• 作业卡在“排队”状态：请验证 RQ 工作进程 ( python -m mcp_waifu_queue.worker ) 是否在单独的终端中运行，并连接到.env中指定的同一 Redis 实例。检查工作进程日志中是否存在错误。
• ConnectionRefusedError (Redis) ：确保您的 Redis 服务器正在运行，并且可以通过.env中指定的REDIS_URL访问。
• MCP 服务器连接问题：确保 MCP 服务器（ uvicorn ... ）正在运行并且您正在连接到正确的主机/端口。

贡献

1. 分叉存储库。
2. 为您的功能或错误修复创建一个新的分支（ git checkout -b feature/your-feature-name ）。
3. 进行更改并提交它们（ git commit -am 'Add some feature' ）。
4. 将您的分支推送到您的分叉存储库（ git push origin feature/your-feature-name ）。
5. 在原始存储库上创建一个新的 Pull 请求。

请遵守项目的编码标准和 linting 规则（ ruff ）。

执照

该项目根据 MIT-0 许可证获得许可 - 有关详细信息，请参阅LICENSE文件。