LLM Wrapper MCP Server

LLM Wrapper MCP Server

“让任何支持 MCP 的 LLM Agent 都能通过 OpenRouter.ai API 与其他任意 LLM 通信或委派任务。”

这是一个 Model Context Protocol（MCP）服务包装器，旨在通过标准化接口，轻松与各种大语言模型（LLM）交互。本项目提供了一个稳健灵活、基于 STDIO 的服务，可处理 LLM 调用、工具执行和结果处理，从而帮助开发者将 LLM 功能集成到应用程序中。

功能

• 实现 Model Context Protocol（MCP）规范，以标准化 LLM 交互。
• 提供基于 STDIO 的服务，通过标准输入／输出来处理 LLM 请求和响应。
• 通过 MCP 协议支持工具调用和结果等高级功能。
• 可通过 API 基础 URL 和模型参数配置，以使用各种 LLM 提供商（例如 OpenRouter、本地模型）。
• 具备可扩展性，可轻松集成新的 LLM 后端。
• 集成 llm-accounting，实现稳健的日志记录、速率限制和审计功能，支持监控远程 LLM 的使用情况和推理成本，并检查查询／响应，以便调试或用于法律目的。

依赖项

本项目依赖以下关键项：

核心依赖项

• pydantic：使用 Python 类型提示进行数据验证和设置管理。
• pydantic-settings：Pydantic 的设置管理功能，用于处理环境变量和配置。
• python-dotenv：从 .env 文件读取键值对，并将其设为环境变量。
• requests：优雅简洁的 Python HTTP 库。
• tiktoken：用于 OpenAI 模型的快速 BPE 分词器。
• llm-accounting：用于稳健的日志记录、速率限制和审计功能。

开发依赖项

• pytest：功能成熟完备的 Python 测试框架。

• black：严格的 Python 代码格式化工具。

• isort：用于按字母顺序排列导入项，并自动按章节和类型分类的 Python 实用工具／库。

• mypy：可选的 Python 静态类型检查器。

• pytest-mock：提供 mocker fixture、便于进行模拟的 pytest 插件。

安装

llm-wrapper-mcp-server 包已发布到 PyPI，可通过 pip 安装：

pip install llm-wrapper-mcp-server

或者，可用于本地开发或从源代码安装：

1. 创建并激活虚拟环境：

python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

2. 安装软件包：

pip install -e .

配置

在项目根目录创建一个 .env 文件，并添加以下变量：

OPENROUTER_API_KEY=your_openrouter_api_key_here

服务默认配置为使用 OpenRouter。API 密钥从 OPENROUTER_API_KEY 环境变量中读取。具体的 LLM 模型和 API 基础 URL 主要通过运行服务时的命令行参数进行配置（见下文）。

未通过 CLI 参数覆盖时的默认设置：

• LLMClient 的 API 基础 URL：https://openrouter.ai/api/v1（可通过 LLM_API_BASE_URL 环境变量或 --llm-api-base-url CLI 参数覆盖）
• LLMClient 的默认模型：perplexity/llama-3.1-sonar-small-128k-online（可通过 --model CLI 参数覆盖）

用法

文字说明：

• Agent 软件通过 MCP 协议（stdin/stdout）与 LLM Wrapper MCP Server 通信。
• LLM Wrapper MCP Server 与 LLM 提供商（例如 OpenRouter.ai）交互，以调用 LLM API 并获取响应。
• 服务还集成了 LLM 记账系统，用于日志记录和审计。
• 主要组件：
  • MCP 通信处理器
  • LLM 客户端
  • 工具执行器
  • LLM 记账集成

运行服务

要运行服务，请执行以下命令：

python -m llm_wrapper_mcp_server [OPTIONS]

例如：

python -m llm_wrapper_mcp_server --model your-org/your-model-name --log-level DEBUG

运行 python -m llm_wrapper_mcp_server --help，查看配置服务的所有可用命令行选项。

此服务以 Model Context Protocol (MCP) STDIO 服务的形式运行，通过标准输入和输出进行通信。它不会为 MCP 通信开放网络端口。

MCP 通信

服务通过 stdin 和 stdout 使用 JSON-RPC 消息进行通信。它支持以下 MCP 方法：

• initialize：握手以建立协议版本和服务功能。
• tools/list：列出可用工具。主服务提供一个 llm_call 工具。
• tools/call：执行指定工具。
• resources/list：列出可用资源（目前没有）。
• resources/templates/list：列出可用资源模板（目前没有）。

llm_call 工具接受 prompt（字符串，必填）和可选的 model（字符串）作为参数；如果指定的模型获准使用，则可按调用覆盖模型。

客户端交互示例（Python）

你可以使用任何支持标准输入/输出通信的语言与 STDIO MCP 服务交互。下面是一个使用 subprocess 模块的 Python 示例：

import subprocess
import json
import time

def send_request(process, request):
    """Sends a JSON-RPC request to the server's stdin."""
    request_str = json.dumps(request) + "\\n"
    process.stdin.write(request_str.encode('utf-8'))
    process.stdin.flush()

def read_response(process):
    """Reads a JSON-RPC response from the server's stdout."""
    line = process.stdout.readline().decode('utf-8').strip()
    if line:
        return json.loads(line)
    return None

if __name__ == "__main__":
    # Start the MCP server as a subprocess
    # Ensure you have the virtual environment activated or the package installed globally
    server_process = subprocess.Popen(
        ["python", "-m", "llm_wrapper_mcp_server"], # Add any CLI args here if needed
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE, # Capture stderr for debugging
        text=False # Use bytes for stdin/stdout
    )

    print("Waiting for server to initialize...")
    # The server sends an initial capabilities message on startup (id: None)
    initial_response = read_response(server_process)
    print(f"Server Initial Response: {json.dumps(initial_response, indent=2)}")

    # 1. Send an 'initialize' request
    initialize_request = {
        "jsonrpc": "2.0",
        "id": "1",
        "method": "initialize",
        "params": {}
    }
    print("\\nSending initialize request...")
    send_request(server_process, initialize_request)
    initialize_response = read_response(server_process)
    print(f"Initialize Response: {json.dumps(initialize_response, indent=2)}")

    # 2. Send a 'tools/call' request to use the 'llm_call' tool
    llm_call_request = {
        "jsonrpc": "2.0",
        "id": "2",
        "method": "tools/call",
        "params": {
            "name": "llm_call",
            "arguments": {
                "prompt": "What is the capital of France?"
                # Optionally add: "model": "another-model/if-allowed"
            }
        }
    }
    print("\\nSending llm_call request...")
    send_request(server_process, llm_call_request)
    llm_call_response = read_response(server_process)
    print(f"LLM Call Response: {json.dumps(llm_call_response, indent=2)}")

    # You can also read stderr for any server logs/errors
    # Note: stderr might block if there's no output, consider using non-blocking reads or threads for real apps
    # stderr_output = server_process.stderr.read().decode('utf-8')
    # if stderr_output:
    #     print("\\nServer Stderr Output:\\n", stderr_output)

    # Terminate the server process
    server_process.terminate()
    server_process.wait(timeout=5) # Wait for process to terminate
    print("\\nServer process terminated.")

开发

如需详细了解项目的目录和文件结构，以及软件开发 Agent 的重要指南，请参阅 AGENTS.md。对于为代码库做出贡献的 Agent 来说，此文档至关重要。

运行测试

此项目使用 pytest 进行测试。

运行所有单元测试：

pytest

集成测试默认处于禁用状态，以避免在正常运行测试时调用外部 API。若要启用并运行集成测试，请使用 integration 标记：

pytest -m integration

安装开发依赖项

安装开发依赖项：

pip install -e ".[dev]"

许可证

MIT 许可证