mcp测试框架
用于测试MCP（模型上下文协议）服务器的pytest插件。
快速开始
pip install mcp-test-frameworkimport pytest
from mcp import StdioServerParameters
from pytest_mcp import assert_tool_exists
@pytest.fixture
def mcp_server():
return StdioServerParameters(
command="python", args=["my_server.py"]
)
async def test_my_tool(mcp_client):
await assert_tool_exists(mcp_client, "my_tool")
result = await mcp_client.call_tool("my_tool", {"arg": "value"})
assert result is not None该插件自动处理服务器生命周期、连接管理，并提供丰富的断言。
特性
模拟MCP客户端
在没有网络开销的情况下测试服务器：
from pytest_mcp import MockMCPClient
async with MockMCPClient(command="python", args=["server.py"]) as client:
tools = await client.list_tools()
result = await client.call_tool("add", {"a": 1, "b": 2})自动注射夹具
定义您的服务器设备并自动连接客户端：
@pytest.fixture
def mcp_server():
return {"command": "python", "args": ["server.py"]}
async def test_tool(mcp_client):
tools = await mcp_client.list_tools()
assert len(tools) > 0丰富的断言
使用为MCP测试设计的描述性断言：
from pytest_mcp import (
assert_tool_exists,
assert_tool_output_matches,
assert_tool_returns_error,
assert_resource_exists,
)
async def test_calculator(mcp_client):
await assert_tool_exists(mcp_client, "add")
result = await mcp_client.call_tool("add", {"a": 2, "b": 3})
await assert_tool_output_matches(result, 5)
await assert_tool_returns_error(
mcp_client, "divide", {"a": 1, "b": 0},
error_message="division by zero"
)
await assert_resource_exists(mcp_client, "config://settings")快照测试
保存并比较测试运行中的工具输出：
async def test_user_data(mcp_client, snapshot):
result = await mcp_client.call_tool("get_user", {"id": 1})
snapshot.assert_match(result, "user_1_response")需要时更新快照：
pytest --mcp-update-snapshots服务器生命周期管理
控制服务器启动和关闭以进行集成测试：
from pytest_mcp import MCPTestServer
async def test_integration():
async with MCPTestServer("python", ["server.py"]) as server:
client = server.get_client()
result = await client.call_tool("hello", {"name": "world"})
await server.restart()API 参考
客户端
模拟MCPClient
MockMCPClient(
server_params: StdioServerParameters | None = None,
*,
command: str | None = None,
args: Sequence[str] | None = None,
env: dict[str, str] | None = None,
)方法：
async list_tools() -> list[Tool]-列出可用工具async call_tool(name: str, arguments: dict) -> CallToolResult-执行工具async list_resources() -> list[Resource]-列出可用资源async read_resource(uri: str) -> ReadResourceResult-阅读资源async get_tool(name: str) -> Tool | None-按名称获取特定工具
夹具
mcp_client-自动注入客户端连接到您的服务器mcp_server-返回服务器参数的用户定义夹具mcp_test_server-具有生命周期控制的高级夹具snapshot-快照测试助手mcp_server_env-服务器的环境变量
断言
# Tool assertions
await assert_tool_exists(client, "tool_name")
await assert_tool_count(client, expected_count)
await assert_tool_output_matches(result, expected_value, partial=False)
await assert_tool_returns_error(client, "tool_name", args, error_message="...")
await assert_tools_have_unique_names(client)
# Schema validation
assert_tool_schema_valid(tool)
# Resource assertions
await assert_resource_exists(client, "resource://uri")
await assert_resource_content_matches(client, "resource://uri", expected_content)快照测试
# JSON snapshots
snapshot.assert_match(data, "snapshot_name")
snapshot.assert_match_json({"key": "value"}, "json_snapshot")
# Text snapshots
snapshot.assert_match_text("output", "text_snapshot")
# Utilities
snapshot.get_snapshot("name")
snapshot.delete_snapshot("name")
snapshot.list_snapshots()服务器管理
async with MCPTestServer(command, args, env) as server:
client = server.get_client()
await server.restart()
await server.wait_for_ready()使用示例
基本计算器服务器
服务器.py：
from mcp.server import Server
from mcp.types import Tool, TextContent
app = Server("calculator")
@app.list_tools()
async def list_tools():
return [
Tool(
name="add",
description="Add two numbers",
inputSchema={
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"},
},
"required": ["a", "b"],
},
)
]
@app.call_tool()
async def call_tool(name, arguments):
if name == "add":
result = arguments["a"] + arguments["b"]
return [TextContent(type="text", text=str(result))]test_server.py：
import pytest
from pytest_mcp import assert_tool_exists, assert_tool_output_matches
@pytest.fixture
def mcp_server():
return {"command": "python", "args": ["server.py"]}
async def test_add(mcp_client):
await assert_tool_exists(mcp_client, "add")
result = await mcp_client.call_tool("add", {"a": 5, "b": 3})
await assert_tool_output_matches(result, "8")高级功能
测试资源：
async def test_resources(mcp_client):
resources = await mcp_client.list_resources()
assert len(resources) > 0
content = await mcp_client.read_resource("config://settings")
assert content is not None错误处理：
async def test_validation(mcp_client):
await assert_tool_returns_error(
mcp_client,
"divide",
{"a": 10, "b": 0},
error_message="Cannot divide by zero"
)快照测试：
async def test_complex_output(mcp_client, snapshot):
result = await mcp_client.call_tool("get_report", {"id": 123})
snapshot.assert_match(result, "report_123")配置
pytest.ini / pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"
markers = [
"mcp: MCP server test (auto-applied)",
"mcp_integration: MCP integration test",
"mcp_slow: Slow MCP test",
]命令行选项
# Set log level
pytest --mcp-log-level=DEBUG
# Set operation timeout
pytest --mcp-timeout=60
# Update snapshots
pytest --mcp-update-snapshots与FastMCP集成
使用 FastMCP:
from fastmcp import FastMCP
from pytest_mcp import MockMCPClient
mcp = FastMCP("My Server")
@mcp.tool()
def greet(name: str) -> str:
return f"Hello, {name}!"
@pytest.fixture
def mcp_server():
return mcp.get_server_params()
async def test_greet(mcp_client):
result = await mcp_client.call_tool("greet", {"name": "Alice"})
await assert_tool_output_matches(result, "Hello, Alice!")贡献
欢迎捐款。开始：
git clone https://github.com/aryanjp1/pytest-mcp.git
cd pytest-mcp
pip install -e ".[dev]"
pytest
black .
ruff check .
mypy src/看 贡献.md 详细指南。
许可证
MIT许可证-请参阅 许可证 文件。
致谢
资源
社区
- -Bug报告和功能请求
- 讨论 -问题和想法