MCP use test

MCP测试框架（基于MCP使用）

基于以下内容构建的测试框架 mcp使用 用于记录和回放MCP工具调用。

中文文档

主要特点

1. mcp使用集成：基于mcp使用构建，用于强大的mcp服务器管理
2. 中间件记录：用于拦截和记录所有工具调用的自定义中间件
3. 呼叫链回放：将调用链保存为JSON，并在没有LLM的情况下回放
4. 自然语言测试：以JSON或INI格式定义数据驱动测试的测试场景
5. 中文语言支持：完全支持使用INI格式的中文自然语言测试用例
6. pytest集成：完全支持pytest和测试生成

为什么是这个框架？

该框架利用 mcp-use的中间件系统 提供：

• 透明记录工具调用
• 无需LLM参与的回放功能
• 易于调试和测试的工作流程
• 数据驱动的测试场景

建筑

mcp_use_test/
├── src/
│   ├── config.py              # Configuration management
│   ├── mcp_manager.py         # MCP server connection manager
│   ├── middleware.py          # Call recording middleware
│   ├── call_chain.py          # Call chain save/replay
│   └── nl_executor.py         # Natural language test executor
├── tests/
│   ├── conftest.py            # pytest fixtures
│   ├── test_config.py         # Configuration tests
│   ├── test_middleware.py     # Middleware tests
│   ├── test_call_chain.py     # Call chain tests
│   └── test_data/
│       └── scenarios.json     # Test scenarios
├── examples/
│   ├── basic_usage.py         # Basic usage example
│   ├── record_and_replay.py   # Record/replay example
│   └── natural_language_test.py  # NL testing example
├── call_chains/               # Saved call chains
├── mcp_config.json           # MCP server configuration
└── requirements.txt          # Python dependencies

安装

# Install dependencies
pip install -r requirements.txt

# Or install in development mode
pip install -e .

配置

编辑 mcp_config.json 要配置MCP服务器，请执行以下操作：

{
  "servers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
      "env": {}
    },
    "git": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-git"],
      "env": {}
    }
  }
}

用法

基本用法

import asyncio
from src.config import Config
from src.mcp_manager import MCPManager
from src.middleware import CallRecorderMiddleware

async def main():
    config = Config.from_file("mcp_config.json")
    middleware = CallRecorderMiddleware(enabled=True)

    async with MCPManager(config) as manager:
        # List available tools
        tools = manager.get_all_tools()
        print(f"Available tools: {len(tools)}")

        # Call a tool with recording
        async def call_func(arguments, **kwargs):
            return await manager.call_tool(
                tool_name="list_directory",
                arguments=arguments,
                server_name="filesystem"
            )

        wrapped_call = middleware.wrap_call(call_func, "list_directory", "filesystem")
        result = await wrapped_call({"path": "."})

        # Print statistics
        middleware.print_summary()

asyncio.run(main())

记录和回放呼叫链

from src.call_chain import CallChain, CallChainManager

# After recording with middleware
chain = CallChain.from_calls(
    calls=middleware.recorded_calls,
    name="my_workflow",
    description="Test workflow"
)

chain_manager = CallChainManager()
await chain_manager.save_chain(chain)

# Later, replay the chain
loaded_chain = await chain_manager.load_chain(chain.id)
results = await chain_manager.replay_chain(loaded_chain, manager)

自然语言测试

JSON格式（英文）

用JSON定义测试场景：

{
  "scenarios": [
    {
      "name": "filesystem_operations",
      "description": "Test filesystem operations",
      "tags": ["filesystem", "basic"],
      "steps": [
        {
          "action": "List files in directory",
          "tool": "list_directory",
          "server": "filesystem",
          "arguments": {"path": "/tmp"},
          "validate": {"not_null": true}
        }
      ]
    }
  ]
}

INI格式（中文自然语言）

使用自然中文以INI格式定义测试场景：

# 中文自然语言测试用例
[scenario_daily_workflow]
name = 日常工作流程测试
description = 模拟用户的日常文件操作工作流程
tags = 日常使用, 工作流程
expected_outcomes = 
    用户能够顺利完成文件管理任务
    所有操作都应该符合预期

step_1 = 1. 查看当前工作目录有哪些文件
step_2 = 2. 创建一个工作笔记文件记录今天的任务
step_2_tool = write_file
step_2_args = path=work_notes.txt,content=今天的工作任务
step_3 = 3. 读取刚才创建的笔记确认内容正确

执行场景：

from src.nl_executor import NaturalLanguageExecutor

executor = NaturalLanguageExecutor(manager, middleware)

# Execute JSON scenarios
results = await executor.execute_scenarios_from_file(
    "tests/test_data/scenarios.json",
    save_chains=True
)

# Execute INI scenarios (Chinese)
results = await executor.execute_scenarios_from_file(
    "test_scenarios.ini",
    save_chains=True
)

# Filter by tags
results = await executor.execute_scenarios_from_file(
    "test_scenarios.ini",
    filter_tags=["基础测试"]
)

运行测试

# Run all tests
pytest

# Run with coverage
pytest --cov=src --cov-report=html

# Run specific test file
pytest tests/test_middleware.py

# Run tests with specific markers
pytest -m "not integration"

示例

这 examples/ 目录包含完整的工作示例：

1. basic_usage.py：基本MCP连接和工具调用
2. record_and_replay.py：录制和回放呼叫链
3. natural_language_test.py：使用自然语言进行数据驱动测试
4. chinese_nl_test.py：INI格式的中文自然语言测试

运行示例：

python examples/basic_usage.py
python examples/record_and_replay.py
python examples/natural_language_test.py
python examples/chinese_nl_test.py

关键组件

MCPManager

管理与多个MCP服务器的连接：

config = Config.from_file("mcp_config.json")
async with MCPManager(config) as manager:
    # Get all tools across all servers
    tools = manager.get_all_tools()

    # Call a tool
    result = await manager.call_tool("tool_name", {"arg": "value"})

CallRecorder中间件

记录工具调用以进行分析和回放：

middleware = CallRecorderMiddleware(enabled=True)
wrapped_call = middleware.wrap_call(func, "tool_name", "server_name")

# Get statistics
stats = middleware.get_statistics()
middleware.print_summary()

呼叫链管理器

管理已保存的呼叫链：

chain_manager = CallChainManager(storage_dir="call_chains")

# Save a chain
await chain_manager.save_chain(chain)

# List all chains
chains = await chain_manager.list_chains()

# Replay a chain
results = await chain_manager.replay_chain(chain, manager)

# Export as pytest test
await chain_manager.export_chain_to_test(chain, "test_output.py")

自然语言执行人

执行数据驱动的测试场景：

executor = NaturalLanguageExecutor(manager, middleware)

# Load and execute scenarios
scenarios = await executor.load_scenarios("scenarios.json")
result = await executor.execute_scenario(scenarios[0])

# Or execute all scenarios from file
results = await executor.execute_scenarios_from_file("scenarios.json")

用例

1.调试和开发

在开发过程中记录工具调用序列，并在不涉及LLM的情况下回放以进行调试：

# During development - record calls
middleware = CallRecorderMiddleware(enabled=True)
# ... perform operations ...

# Save the chain
chain = CallChain.from_calls(middleware.recorded_calls, "debug_session")
await chain_manager.save_chain(chain)

# Later - replay for debugging
loaded_chain = await chain_manager.load_chain("debug_session_...")
await chain_manager.replay_chain(loaded_chain, manager)

2.自动化测试

用自然语言编写测试场景并自动执行：

{
  "scenarios": [
    {
      "name": "user_workflow",
      "steps": [
        {"action": "Create file", "tool": "write_file", ...},
        {"action": "Read file", "tool": "read_file", ...},
        {"action": "Delete file", "tool": "delete_file", ...}
      ]
    }
  ]
}

3.集成测试

跨多个MCP服务器的测试工作流程：

# Test a workflow using filesystem, git, and database servers
executor = NaturalLanguageExecutor(manager, middleware)
results = await executor.execute_scenarios_from_file("integration_tests.json")

4.回归测试

将记录的呼叫链导出为用于回归测试的pytest测试：

await chain_manager.export_chain_to_test(chain, "tests/test_regression.py")

益处

1. LLM独立重播：一旦记录，呼叫链可以在没有LLM参与的情况下重放
2. 数据驱动测试：以自然语言JSON或INI格式定义测试
3. 中文语言支持：使用INI格式用自然中文编写测试用例
4. 可观测性：通过计时和错误跟踪全面了解工具调用
5. 调试：通过调用链回放轻松调试
6. 测试自动化：与CI/CD管道的pytest集成
7. 多服务器：跨多个MCP服务器的测试工作流

贡献

欢迎投稿！请随时提交问题或拉取请求。

许可证

MIT许可证