MCP Waifu Queue（双子版）

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

目录

• 特征

• 建筑学

• 先决条件

• 安装

• 配置

• 运行服务

• MCP API

• 测试

• 故障排除

• 贡献

• 执照

Related MCP server: Selector AI FastMCP

特征

• 使用 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 ）：

   使用

   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

   使用

   # 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 install -e .[test] # Installs package in editable mode with test extras

   使用

   # 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

   （将

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 调用，具体取决于它们的实现。

故障排除

• 错误： ：请确保您已在主目录中创建~/.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文件。