FastMCP测试工具:自动化测试的基础设施
FastMCP测试工具:自动化测试的基础设施
【免费下载链接】fastmcp The fast, Pythonic way to build Model Context Protocol servers 🚀 项目地址: 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
实现了零开销的内存连接。
基础内存测试示例
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
测试最佳实践
测试组织结构
测试配置管理
# 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的测试工具,你可以:
- 零配置开始测试:无需服务器部署即可运行完整测试套件
- 获得即时反馈:内存测试使测试执行时间从秒级降到毫秒级
- 全面覆盖:从单元测试到端到端测试的完整支持
- 无缝调试:在测试代码和服务器代码中都可以设置断点
- 持续集成友好:简单的配置即可集成到CI/CD流水线
FastMCP让MCP服务器测试从复杂的基础设施挑战变成了简单的代码质量保障,真正实现了\"测试即代码\"的理念。
【免费下载链接】fastmcp The fast, Pythonic way to build Model Context Protocol servers 🚀 项目地址: https://gitcode.com/GitHub_Trending/fa/fastmcp
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考