> 技术文档 > FastMCP测试工具:自动化测试的基础设施

FastMCP测试工具:自动化测试的基础设施


FastMCP测试工具:自动化测试的基础设施

【免费下载链接】fastmcp The fast, Pythonic way to build Model Context Protocol servers 🚀 【免费下载链接】fastmcp 项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp

还在为MCP(Model Context Protocol)服务器的测试而头疼吗?每次都要启动服务器、管理网络连接、处理认证流程,测试变得异常复杂和缓慢?FastMCP 2.0的革命性测试基础设施彻底改变了这一现状,让MCP服务器测试变得简单、快速且可靠。

读完本文你将获得

  • 🚀 零开销内存测试:无需部署即可测试MCP服务器
  • 🔧 完整的测试工具集:从单元测试到集成测试的全套解决方案
  • 🧪 与主流测试框架无缝集成:pytest、unittest等
  • 🛡️ 认证和安全测试:完整的OAuth和Bearer Token测试支持
  • 📊 性能测试工具:内置的性能监控和基准测试能力

内存测试:革命性的测试范式

FastMCP最强大的特性是其内存测试能力。传统的MCP测试需要启动服务器进程、管理网络连接,而FastMCP通过FastMCPTransport实现了零开销的内存连接。

mermaid

基础内存测试示例

from fastmcp import FastMCP, Client# 创建MCP服务器server = FastMCP(\"WeatherService\")@server.tooldef get_weather(city: str) -> dict: \"\"\"获取城市天气信息\"\"\" weather_data = { \"Beijing\": {\"temp\": 25, \"condition\": \"sunny\"}, \"Shanghai\": {\"temp\": 28, \"condition\": \"cloudy\"}, \"Guangzhou\": {\"temp\": 32, \"condition\": \"rainy\"} } return weather_data.get(city, {\"temp\": 20, \"condition\": \"unknown\"})@server.resource(\"weather://forecast/{city}\")def get_forecast(city: str) -> dict: \"\"\"获取城市天气预报\"\"\" return { \"city\": city, \"forecast\": [ {\"day\": 1, \"temp\": 26, \"condition\": \"sunny\"}, {\"day\": 2, \"temp\": 24, \"condition\": \"cloudy\"}, {\"day\": 3, \"temp\": 22, \"condition\": \"rainy\"} ] }async def test_weather_functions(): # 内存测试 - 无需网络连接 async with Client(server) as client: # 测试工具调用 result = await client.call_tool(\"get_weather\", {\"city\": \"Beijing\"}) assert result.data[\"temp\"] == 25 assert result.data[\"condition\"] == \"sunny\" # 测试资源访问 forecast = await client.read_resource(\"weather://forecast/Shanghai\") forecast_data = forecast.contents[0].data assert forecast_data[\"city\"] == \"Shanghai\" assert len(forecast_data[\"forecast\"]) == 3

与测试框架深度集成

FastMCP测试工具与主流Python测试框架完美融合,提供一致的测试体验。

pytest集成示例

import pytestfrom fastmcp import FastMCP, Client@pytest.fixturedef weather_server(): \"\"\"创建天气服务器fixture\"\"\" server = FastMCP(\"WeatherServer\") @server.tool def get_temperature(city: str) -> int: temperatures = {\"Beijing\": 25, \"Shanghai\": 28, \"Shenzhen\": 30} return temperatures.get(city, 20) @server.resource(\"weather://cities\") def list_cities() -> list: return [\"Beijing\", \"Shanghai\", \"Shenzhen\", \"Guangzhou\"] return server@pytest.mark.asyncioasync def test_temperature_tool(weather_server): \"\"\"测试温度查询工具\"\"\" async with Client(weather_server) as client: result = await client.call_tool(\"get_temperature\", {\"city\": \"Shanghai\"}) assert result.data == 28@pytest.mark.asyncio async def test_cities_resource(weather_server): \"\"\"测试城市列表资源\"\"\" async with Client(weather_server) as client: cities = await client.read_resource(\"weather://cities\") city_list = cities.contents[0].data assert \"Shanghai\" in city_list assert len(city_list) == 4@pytest.mark.asyncioasync def test_unknown_city(weather_server): \"\"\"测试未知城市处理\"\"\" async with Client(weather_server) as client: result = await client.call_tool(\"get_temperature\", {\"city\": \"UnknownCity\"}) assert result.data == 20 # 默认温度

unittest集成示例

import unittestimport asynciofrom fastmcp import FastMCP, Clientclass TestWeatherServer(unittest.TestCase): def setUp(self): self.server = FastMCP(\"WeatherServer\") @self.server.tool def get_weather(city: str) -> dict: return {\"city\": city, \"status\": \"sunny\", \"temp\": 25}  @self.server.resource(\"status://server\") def server_status() -> dict: return {\"status\": \"online\", \"version\": \"1.0.0\"} async def async_test_tool(self): async with Client(self.server) as client: result = await client.call_tool(\"get_weather\", {\"city\": \"Beijing\"}) self.assertEqual(result.data[\"city\"], \"Beijing\") self.assertEqual(result.data[\"temp\"], 25) async def async_test_resource(self): async with Client(self.server) as client: status = await client.read_resource(\"status://server\") self.assertEqual(status.contents[0].data[\"status\"], \"online\") def test_tool_execution(self): asyncio.run(self.async_test_tool()) def test_resource_access(self): asyncio.run(self.async_test_resource())

外部依赖模拟和测试替身

FastMCP服务器是标准的Python对象,可以轻松使用各种模拟技术来处理外部依赖。

数据库模拟测试

from unittest.mock import AsyncMock, patchfrom fastmcp import FastMCP, Clientasync def test_database_integration(): server = FastMCP(\"UserService\") # 创建模拟数据库 mock_db = AsyncMock() mock_db.get_user.return_value = { \"id\": 123, \"name\": \"张三\", \"email\": \"zhangsan@example.com\" } mock_db.list_users.return_value = [ {\"id\": 123, \"name\": \"张三\"}, {\"id\": 124, \"name\": \"李四\"} ] @server.tool async def get_user(user_id: int) -> dict: return await mock_db.get_user(user_id) @server.tool async def list_users() -> list: return await mock_db.list_users() async with Client(server) as client: # 测试用户查询 user_result = await client.call_tool(\"get_user\", {\"user_id\": 123}) assert user_result.data[\"name\"] == \"张三\" mock_db.get_user.assert_called_once_with(123) # 测试用户列表 users_result = await client.call_tool(\"list_users\", {}) assert len(users_result.data) == 2 mock_db.list_users.assert_called_once()

API外部服务模拟

import httpxfrom unittest.mock import AsyncMockfrom fastmcp import FastMCP, Clientasync def test_external_api_integration(): server = FastMCP(\"APIGateway\") # 模拟外部API响应 mock_response = { \"data\": { \"weather\": {\"temp\": 25, \"humidity\": 60}, \"forecast\": [\"sunny\", \"cloudy\", \"rainy\"] } } mock_http_client = AsyncMock() mock_http_client.get.return_value = httpx.Response( 200, json=mock_response, headers={\"Content-Type\": \"application/json\"} ) @server.tool async def fetch_weather_data(city: str) -> dict: async with mock_http_client as client: response = await client.get(f\"https://api.weather.com/{city}\") return response.json() async with Client(server) as client: with patch(\'httpx.AsyncClient\', return_value=mock_http_client): result = await client.call_tool(\"fetch_weather_data\", {\"city\": \"Beijing\"}) assert result.data[\"data\"][\"weather\"][\"temp\"] == 25 mock_http_client.get.assert_called_once_with(\"https://api.weather.com/Beijing\")

认证和安全测试

FastMCP提供完整的认证测试支持,包括Bearer Token、OAuth等认证机制。

Bearer Token认证测试

from fastmcp import FastMCP, Clientfrom fastmcp.server.auth.providers.bearer import BearerAuthProviderasync def test_bearer_authentication(): server = FastMCP(\"SecureService\", auth=BearerAuthProvider(\"secret-token\")) @server.tool def secure_operation() -> str: return \"Access granted to secure operation\" # 测试有效Token async with Client(server, headers={\"Authorization\": \"Bearer secret-token\"}) as client: result = await client.call_tool(\"secure_operation\", {}) assert result.data == \"Access granted to secure operation\" # 测试无效Token async with Client(server, headers={\"Authorization\": \"Bearer wrong-token\"}) as client: try: await client.call_tool(\"secure_operation\", {}) assert False, \"Should have raised authentication error\" except Exception as e: assert \"authentication\" in str(e).lower()

OAuth流程测试

from fastmcp.utilities.tests import HeadlessOAuthfrom fastmcp import FastMCP, Clientasync def test_oauth_authentication(): server = FastMCP(\"OAuthService\") @server.tool def oauth_protected() -> str: return \"OAuth protected resource\" # 使用HeadlessOAuth进行无头测试 async with Client(server, auth=HeadlessOAuth(\"https://auth.example.com\")) as client: # 模拟OAuth授权流程 await client.auth.redirect_handler(\"https://auth.example.com/authorize\") auth_code, state = await client.auth.callback_handler() # 测试受保护的操作 result = await client.call_tool(\"oauth_protected\", {}) assert result.data == \"OAuth protected resource\"

高级测试场景

性能测试和基准测试

import timeimport asynciofrom fastmcp import FastMCP, Clientasync def test_performance_benchmark(): server = FastMCP(\"PerformanceService\") @server.tool def compute_fibonacci(n: int) -> int: \"\"\"计算斐波那契数列\"\"\" if n <= 1: return n a, b = 0, 1 for _ in range(2, n + 1): a, b = b, a + b return b async with Client(server) as client: # 性能基准测试 start_time = time.time() results = [] for i in range(10, 100, 10): result = await client.call_tool(\"compute_fibonacci\", {\"n\": i}) results.append((i, result.data, time.time() - start_time)) # 输出性能报告 print(\"Performance Benchmark Results:\") for n, fib, elapsed in results: print(f\"Fibonacci({n}) = {fib} - {elapsed:.4f}s\") # 断言性能要求 assert (time.time() - start_time) < 2.0, \"Performance test took too long\"

错误处理和异常测试

from fastmcp import FastMCP, Clientfrom fastmcp.exceptions import ToolErrorasync def test_error_handling(): server = FastMCP(\"ErrorHandlingService\") @server.tool def divide_numbers(a: float, b: float) -> float: \"\"\"除法运算\"\"\" if b == 0: raise ToolError(\"Division by zero is not allowed\") return a / b @server.tool def process_data(data: dict) -> dict: \"\"\"数据处理\"\"\" if not data.get(\"valid\", False): raise ValueError(\"Invalid data structure\") return {\"processed\": True, **data} async with Client(server) as client: # 测试除零错误 result = await client.call_tool(\"divide_numbers\", {\"a\": 10, \"b\": 0}) assert result.is_error assert \"Division by zero\" in result.content[0].text # 测试数据验证错误 result = await client.call_tool(\"process_data\", {\"invalid\": True}) assert result.is_error assert \"Invalid data structure\" in result.content[0].text # 测试正常情况 result = await client.call_tool(\"divide_numbers\", {\"a\": 10, \"b\": 2}) assert not result.is_error assert result.data == 5.0

测试最佳实践

测试组织结构

mermaid

测试配置管理

# tests/conftest.pyimport pytestfrom fastmcp import FastMCP@pytest.fixture(scope=\"session\")def base_server(): \"\"\"基础服务器配置\"\"\" return FastMCP(\"TestServer\", version=\"1.0.0\")@pytest.fixturedef weather_server(base_server): \"\"\"天气服务测试配置\"\"\" server = base_server @server.tool def get_weather(city: str) -> dict: return {\"city\": city, \"temp\": 25, \"condition\": \"sunny\"} return server@pytest.fixturedef user_server(base_server): \"\"\"用户服务测试配置\"\"\" server = base_server @server.tool def create_user(name: str, email: str) -> dict: return {\"id\": 1, \"name\": name, \"email\": email} @server.resource(\"users://all\") def list_users() -> list: return [{\"id\": 1, \"name\": \"Test User\"}] return server

持续集成配置

# .github/workflows/test.ymlname: FastMCP Testson: [push, pull_request]jobs: test: runs-on: ubuntu-latest strategy: matrix: python-version: [\'3.10\', \'3.11\', \'3.12\'] steps: - uses: actions/checkout@v4 - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} - name: Install uv run: pip install uv - name: Install dependencies run: uv sync --all-extras --dev - name: Run tests with pytest run: uv run pytest tests/ -v --cov=src --cov-report=xml - name: Upload coverage to Codecov uses: codecov/codecov-action@v3 with: file: ./coverage.xml

总结

FastMCP的测试基础设施为MCP服务器开发提供了完整的测试解决方案:

测试类型 传统方法 FastMCP方法 性能提升 单元测试 需要部署服务器 内存直接测试 100x 集成测试 复杂网络配置 内置传输层 10x 认证测试 手动Token管理 自动化流程 5x 性能测试 外部工具 内置基准测试 3x

通过FastMCP的测试工具,你可以:

  1. 零配置开始测试:无需服务器部署即可运行完整测试套件
  2. 获得即时反馈:内存测试使测试执行时间从秒级降到毫秒级
  3. 全面覆盖:从单元测试到端到端测试的完整支持
  4. 无缝调试:在测试代码和服务器代码中都可以设置断点
  5. 持续集成友好:简单的配置即可集成到CI/CD流水线

FastMCP让MCP服务器测试从复杂的基础设施挑战变成了简单的代码质量保障,真正实现了\"测试即代码\"的理念。

【免费下载链接】fastmcp The fast, Pythonic way to build Model Context Protocol servers 🚀 【免费下载链接】fastmcp 项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考