mcpium (Realdnchka)

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 参考

步骤动作

修饰符

断言

配置

使用 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 Python SDK（MCP Python软件开发工具包）

______________________________________________________________________

为MCP社区倾心打造