How to test MCP? What are the testing strategies and best practices?

Testing strategies for MCP are critical for ensuring system quality and reliability. Here are detailed testing methods and best practices:

Testing Hierarchy

MCP testing should cover the following levels:

1. Unit Testing: Test individual functions and components
2. Integration Testing: Test interactions between components
3. End-to-End Testing: Test complete user scenarios
4. Performance Testing: Test system performance and scalability
5. Security Testing: Test security vulnerabilities and protection mechanisms

1. Unit Testing

python
import pytest
from unittest.mock import Mock, AsyncMock
from mcp.server import Server

class TestMCPTools:
    @pytest.fixture
    def server(self):
        """Create test server instance"""
        return Server("test-server")
    
    @pytest.mark.asyncio
    async def test_tool_registration(self, server):
        """Test tool registration"""
        @server.tool(
            name="test_tool",
            description="Test tool"
        )
        async def test_tool(param: str) -> str:
            return f"Result: {param}"
        
        # Verify tool is registered
        tools = await server.list_tools()
        assert any(t["name"] == "test_tool" for t in tools)
    
    @pytest.mark.asyncio
    async def test_tool_execution(self, server):
        """Test tool execution"""
        @server.tool(
            name="calculate",
            description="Calculation tool"
        )
        async def calculate(a: int, b: int) -> int:
            return a + b
        
        # Execute tool
        result = await server.call_tool("calculate", {"a": 2, "b": 3})
        assert result == 5
    
    @pytest.mark.asyncio
    async def test_parameter_validation(self, server):
        """Test parameter validation"""
        @server.tool(
            name="validate",
            description="Parameter validation tool",
            inputSchema={
                "type": "object",
                "properties": {
                    "email": {
                        "type": "string",
                        "format": "email"
                    }
                },
                "required": ["email"]
            }
        )
        async def validate(email: str) -> str:
            return f"Valid: {email}"
        
        # Test valid parameters
        result = await server.call_tool(
            "validate",
            {"email": "test@example.com"}
        )
        assert "Valid" in result
        
        # Test invalid parameters
        with pytest.raises(ValueError):
            await server.call_tool("validate", {"email": "invalid"})

2. Integration Testing

python
class TestMCPIntegration:
    @pytest.mark.asyncio
    async def test_client_server_communication(self):
        """Test client-server communication"""
        from mcp.client import Client
        from mcp.server import Server
        
        # Create server
        server = Server("integration-test-server")
        
        @server.tool(name="echo", description="Echo tool")
        async def echo(message: str) -> str:
            return message
        
        # Start server
        await server.start()
        
        try:
            # Create client
            client = Client("http://localhost:8000")
            
            # Test communication
            result = await client.call_tool("echo", {"message": "Hello"})
            assert result == "Hello"
        
        finally:
            await server.stop()
    
    @pytest.mark.asyncio
    async def test_resource_access(self):
        """Test resource access"""
        server = Server("resource-test-server")
        
        @server.resource(
            uri="file:///test.txt",
            name="Test File",
            description="Test resource"
        )
        async def test_resource() -> str:
            return "Test content"
        
        await server.start()
        
        try:
            client = Client("http://localhost:8000")
            
            # Read resource
            content = await client.read_resource("file:///test.txt")
            assert content == "Test content"
        
        finally:
            await server.stop()

3. End-to-End Testing

python
class TestMCPEndToEnd:
    @pytest.mark.asyncio
    async def test_complete_workflow(self):
        """Test complete workflow"""
        # Simulate user scenario: query database and generate report
        
        server = Server("e2e-test-server")
        
        @server.tool(name="query_db", description="Query database")
        async def query_db(query: str) -> list:
            return [{"id": 1, "name": "Test"}]
        
        @server.tool(name="generate_report", description="Generate report")
        async def generate_report(data: list) -> str:
            return f"Report: {len(data)} items"
        
        await server.start()
        
        try:
            client = Client("http://localhost:8000")
            
            # Execute workflow
            data = await client.call_tool("query_db", {"query": "SELECT *"})
            report = await client.call_tool("generate_report", {"data": data})
            
            assert "1 items" in report
        
        finally:
            await server.stop()

4. Performance Testing

python
import asyncio
import time
from locust import HttpUser, task, between

class MCPPerformanceTest(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        """Initialization at test start"""
        self.client = Client(self.host)
    
    @task
    def tool_call_performance(self):
        """Test tool call performance"""
        start_time = time.time()
        
        result = self.client.call_tool("test_tool", {"param": "value"})
        
        elapsed = time.time() - start_time
        
        # Assert response time
        assert elapsed < 1.0, f"Response time too long: {elapsed}s"
    
    @task
    def concurrent_requests(self):
        """Test concurrent requests"""
        async def make_request():
            return self.client.call_tool("test_tool", {"param": "value"})
        
        # Execute 10 requests concurrently
        tasks = [make_request() for _ in range(10)]
        results = asyncio.run(asyncio.gather(*tasks))
        
        # Verify all requests succeeded
        assert all(results)

5. Security Testing

python
class TestMCPSecurity:
    @pytest.mark.asyncio
    async def test_authentication(self):
        """Test authentication mechanism"""
        server = Server("security-test-server")
        
        # Configure authentication
        server.set_authenticator(lambda token: token == "valid-token")
        
        @server.tool(name="secure_tool", description="Secure tool")
        async def secure_tool() -> str:
            return "Secure data"
        
        await server.start()
        
        try:
            # Test valid token
            client = Client("http://localhost:8000", token="valid-token")
            result = await client.call_tool("secure_tool", {})
            assert result == "Secure data"
            
            # Test invalid token
            client_invalid = Client("http://localhost:8000", token="invalid-token")
            with pytest.raises(AuthenticationError):
                await client_invalid.call_tool("secure_tool", {})
        
        finally:
            await server.stop()
    
    @pytest.mark.asyncio
    async def test_sql_injection_prevention(self):
        """Test SQL injection prevention"""
        server = Server("sql-injection-test-server")
        
        @server.tool(name="query", description="Query tool")
        async def query(sql: str) -> list:
            # Should use parameterized queries
            return execute_safe_query(sql)
        
        # Test SQL injection attempt
        malicious_sql = "SELECT * FROM users WHERE '1'='1'"
        result = await server.call_tool("query", {"sql": malicious_sql})
        
        # Verify injection is blocked
        assert result == []
    
    @pytest.mark.asyncio
    async def test_rate_limiting(self):
        """Test rate limiting"""
        server = Server("rate-limit-test-server")
        
        # Configure rate limiting
        server.set_rate_limit(max_requests=10, window=60)
        
        @server.tool(name="limited_tool", description="Rate-limited tool")
        async def limited_tool() -> str:
            return "Success"
        
        # Send multiple requests rapidly
        for i in range(15):
            try:
                await server.call_tool("limited_tool", {})
            except RateLimitError:
                # Expected rate limit error
                assert i >= 10
                break
        else:
            pytest.fail("Rate limit not triggered")

6. Mocks and Stubs

python
from unittest.mock import Mock, patch

class TestMCPWithMocks:
    @pytest.mark.asyncio
    async def test_with_external_dependency_mock(self):
        """Test external dependency using Mock"""
        server = Server("mock-test-server")
        
        @server.tool(name="fetch_data", description="Fetch data")
        async def fetch_data(url: str) -> dict:
            # Mock external API call
            with patch('requests.get') as mock_get:
                mock_get.return_value.json.return_value = {
                    "data": "mocked"
                }
                
                response = requests.get(url)
                return response.json()
        
        result = await server.call_tool(
            "fetch_data",
            {"url": "http://api.example.com"}
        )
        
        assert result == {"data": "mocked"}
        mock_get.assert_called_once()

7. Test Coverage

python
# Use pytest-cov to generate coverage reports
# Run command: pytest --cov=mcp --cov-report=html

class TestCoverage:
    @pytest.mark.asyncio
    async def test_all_code_paths(self):
        """Test all code paths"""
        server = Server("coverage-test-server")
        
        @server.tool(name="complex_tool", description="Complex tool")
        async def complex_tool(condition: bool) -> str:
            if condition:
                return "Branch A"
            else:
                return "Branch B"
        
        # Test all branches
        result_a = await server.call_tool("complex_tool", {"condition": True})
        assert result_a == "Branch A"
        
        result_b = await server.call_tool("complex_tool", {"condition": False})
        assert result_b == "Branch B"

Best Practices:

1. Test Pyramid: Many unit tests, moderate integration tests, few end-to-end tests
2. Independence: Each test should run independently without depending on other tests
3. Repeatability: Test results should be repeatable and not affected by environmental factors
4. Fast Feedback: Unit tests should execute quickly to provide fast feedback
5. Continuous Integration: Integrate testing into CI/CD pipelines
6. Coverage Goals: Set reasonable code coverage targets (e.g., 80%)

Through comprehensive testing strategies, you can ensure the quality and reliability of MCP systems.