AI导哈

MCP服务器测试框架

一个针对MCP服务器的现代化、流畅API测试框架，支持链式语法编写确定性测试。

  • by realdnchka
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 realdnchka/mcpium

使用说明

mcpium（注：此词并非标准英文或中文词汇，可能是特定领域或虚构的术语，直接音译为中文，实际含义需根据上下文确定）

针对MCP服务器的Selenium类测试框架

一个现代、流畅的API测试框架，用于 模型上下文协议 服务器。使用简洁、链式语法编写确定性测试。

特点/特性

  • 🔗（这个符号在中文里通常没有直接对应的翻译，它在英文中常用来表示链接或连接，所以可以翻译为“链接”或根据上下文意译为“相关链接”、“点击这里”等，如果单独使用且没有上下文，一般保留为“🔗”或解释为“链接符号”。） 流畅的API - 美观的链式方法调用，使测试更易读
  • 🔄（循环、重复） 重试与超时 - 内置的对不稳定操作的容错能力
  • 🎯（目标） 丰富的断言 - 15+种断言方法，实现全面验证
  • 🛠️（工具或螺丝刀等工具的符号，可具体翻译为“工具”或根据上下文翻译为“螺丝刀”等） 全面支持MCP（可能指某种特定技术、协议或系统，具体需根据上下文确定） - 工具、提示和资源
  • 📝 类型安全 - 完整的类型提示，以支持集成开发环境（IDE）
  • 🚀 火箭/快速上升/飞速发展（根据上下文可灵活翻译） 异步原生（或异步本地） - 基于现代Python的asyncio构建
  • 📊 表格/数据图表 自动记录 - 自动计时和状态跟踪

快速入门

安装

poetry add mcpium
# or
pip install mcpium

基本示例

import asyncio
from mcpium.core.scenario import Scenario

async def main():
    async with Scenario("My first test") as s:
        # Simple fluent chain
        await s.step("List available tools").list_tools().expect().to_succeed()

        # Call a tool and verify result
        await s.step("Call tool").call_tool("get_user", id=123).expect().to_equal({"name": "Alice"})

if __name__ == "__main__":
    asyncio.run(main())

高级用法

重试与超时

async with Scenario("Resilient API Test") as s:
    await s.step("Call flaky API") \
        .call_tool("external_service", endpoint="/data") \
        .with_retry(max_attempts=5, backoff=2.0) \
        .with_timeout(30.0) \
        .expect().to_succeed()

使用提示进行工作

async with Scenario("Prompt Test") as s:
    # List available prompts
    await s.step("List prompts").list_prompts().expect().not_empty()

    # Use a prompt with arguments
    await s.step("Generate greeting") \
        .get_prompt("welcome", {"user": "Alice", "lang": "en"}) \
        .expect().to_match_regex(r"Hello.*Alice")

资源测试

async with Scenario("Resource Test") as s:
    # List available resources
    await s.step("List resources") \
        .list_resources() \
        .expect().to_have_length(5)

    # Read and validate resource
    await s.step("Read config") \
        .read_resource("file://config.json") \
        .expect().to_be_type(dict) \
        .has_key("version")

复杂断言

async with Scenario("Advanced Validation") as s:
    await s.step("Process data") \
        .call_tool("processor", data=payload) \
        .expect() \
        .to_succeed() \
        .to_be_type(dict) \
        .has_key("results") \
        .to_have_length(10) \
        .to_be_between(5, 20)

API 参考

步骤动作

方法 描述
list_tools() 列出所有可用的MCP工具
call_tool(name, **kwargs) 调用带有参数的MCP工具
validate(tool_name, **kwargs) 根据工具模式验证参数
list_prompts() 列出所有可用提示
get_prompt(name, arguments) 获取带有模板参数的提示
list_resources() 列出所有可用资源
read_resource(uri) 通过URI读取资源

修饰符

方法 描述
.with_retry(max_attempts, backoff) 对失败的操作进行指数退避重试
.with_timeout(seconds) 设置操作超时
.expect() 开始断言链

断言

方法 描述
.to_succeed() 断言操作成功
.to_fail() 断言操作失败
.to_equal(value) 断言精确相等
.to_contain(item) 断言包含项（列表/字典/字符串）
.has_key(key) 断言字典包含键
.has_value(value) 断言字典中有该值
.not_empty() 断言结果不为空
.to_match_regex(pattern) 断言匹配正则表达式模式
.to_have_length(n) 断言长度等于 n
.to_be_greater_than(n) 断言值 > n
.to_be_less_than(n) 断言值 < n
.to_be_between(min, max) 断言最小值 ≤ 值 ≤ 最大值
.to_be_one_of(values) 列表中的断言值
.to_be_type(type) 断言特定类型
.to_have_error(substring) 断言错误包含子字符串

配置

使用 TOML 配置文件

创建一个 mcpium.toml 在项目根目录中的文件：

[mcpium]
server_uri = "ws://localhost:8765"
default_timeout = 30.0  # Default timeout in seconds for all operations

程序化配置

from mcpium.core.config import Config
from mcpium.core.driver import MCPDriver
from mcpium.core.scenario import Scenario

# Custom config file
config = Config(config_path="custom_config.toml")

# Driver with custom default timeout
driver = MCPDriver(default_timeout=60.0)

# Scenario with custom default timeout (overrides driver and config)
scenario = Scenario("Custom test", driver=driver, default_timeout=10.0)

超时优先级（从高到低）：

  1. .with_timeout(seconds) 在特定步骤上
  2. Scenario(default_timeout=...)
  3. MCPDriver(default_timeout=...)
  4. mcpium.toml 配置 default_timeout
  5. 无超时（无限期等待）

与 pytest 的集成

import pytest
from mcpium.core.scenario import Scenario

@pytest.fixture
async def scenario():
    async with Scenario("Test") as s:
        yield s

async def test_mcp_server(scenario):
    await scenario.step("Test tool").call_tool("test").expect().to_succeed()

示例

示例1：端到端MCP服务器测试

async with Scenario("E2E Server Test") as s:
    # Verify server capabilities
    await s.step("Check tools").list_tools().expect().not_empty()
    await s.step("Check prompts").list_prompts().expect().not_empty()
    await s.step("Check resources").list_resources().expect().not_empty()

    # Test tool execution
    result = await s.step("Execute tool") \
        .call_tool("calculate", expression="2+2") \
        .with_timeout(5.0)

    result.expect().to_equal(4)

示例2：验证测试

async with Scenario("Schema Validation") as s:
    # Valid arguments
    await s.step("Valid call") \
        .validate("create_user", name="Alice", age=30) \
        .expect().to_succeed()

    # Invalid arguments
    await s.step("Invalid call") \
        .validate("create_user", age="not a number") \
        .expect().to_fail()

示例3：性能测试

async with Scenario("Performance Test") as s:
    import time

    start = time.time()
    await s.step("Fast operation") \
        .call_tool("quick_task") \
        .with_timeout(1.0) \
        .expect().to_succeed()

    duration = time.time() - start
    assert duration < 1.0, f"Operation too slow: {duration}s"

贡献；做出贡献

欢迎贡献！请随时提交拉取请求。

许可证

MIT 许可证 - 请参阅 许可证 详情请参阅文件。

链接

为MCP社区倾心打造

相关 MCP