MCP Tool 单元测试框架：mock LLM 调用的最佳实践

En tant qu'ingénieur qui a passé plus de deux ans à construire des pipelines d'intégration IA, je peux vous confirmer une vérité que personne ne vous dit dans les tutoriels officiels : les tests unitaires avec des appels LLM réels sont un cauchemar logistique et financier. Chaque exécution de test qui frappe une API externe, c'est de l'argent parti, de la latence subie, et une fragilité intégrée dans votre CI/CD. Aujourd'hui, je vais vous montrer comment mocker proprement vos appels LLM avec le framework MCP Tool, en utilisant HolySheep AI comme基础设施 de référence — parce que PERSONNE ne devrait payer $15 le token pour faire tourner des tests.

Tableau comparatif : HolySheep vs API officielle vs services relais

Comme vous pouvez le voir, HolySheep AI offre une compatibilité complète avec le SDK OpenAI sur https://api.holysheep.ai/v1, tout en proposant des tarifs jusqu'à 85% inférieurs. C'est la solution idéale pour les environnements de test où vous voulez la fidélité d'une vraie API sans la facture associée.

Pourquoi mocker les appels LLM dans vos tests unitaires

Dans ma pratique quotidienne, j'ai vu des équipes laisser leurs tests frapper des API réelles "juste pour être sûr". C'est une erreur coûteuse à plusieurs niveaux :

• Coût financier : Un projet avec 500 tests utilisant GPT-4o mini (à $0.15/1M tokens) peut facilement générer $50-200/mois en appels de test.
• Instabilité des tests : Les LLM sont non-déterministes par nature. Un test qui vérifie "la réponse contient X" peut échouer parce que le modèle a changé son style.
• Latence CI/CD : 500 appels × 200ms = 100 secondes MINIMUM ajoutées à votre pipeline.
• Rate limiting : Les API ont des limites. Vos tests peuvent déclencher des blocks sans prévenir.
• Confidentialité : Vos prompts de test contiennent parfois des données sensibles qui ne devraient pas quitter vos serveurs.

Architecture du MCP Tool pour le mocking

Le framework MCP (Model Context Protocol) Tool fournit une abstraction élégante pour interfacer vos outils avec les LLMs. Pour les tests, nous allons créer un mock server qui simule exactement le comportement du serveur HolySheep sans faire d'appels réseau réels.

Implémentation du mock LLM avec le framework MCP Tool

1. Configuration de base du projet

# requirements.txt
pytest>=8.0.0
pytest-asyncio>=0.23.0
mcp>=1.0.0
openai>=1.12.0
httpx>=0.27.0
responses>=0.25.0
fakeredis>=2.21.0

Installation
pip install -r requirements.txt

2. Mock Server MCP Tool

"""
mock_mcp_server.py
Mock du serveur MCP Tool pour les tests unitaires
Compatible avec l'API HolySheep AI sur https://api.holysheep.ai/v1
"""

import json
import asyncio
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from mcp.server import Server
from mcp.types import Tool, CallToolResult, TextContent

@dataclass
class MockLLMResponse:
    """Structure pour définir les réponses mockées"""
    model: str
    content: str
    finish_reason: str = "stop"
    latency_ms: int = 45

class MCPMockServer:
    """
    Serveur mock implémentant le protocole MCP Tool.
    Simule le comportement de HolySheep AI sans appels réseau.
    """
    
    def __init__(self):
        self.server = Server("mcp-mock-server")
        self.responses: Dict[str, MockLLMResponse] = {}
        self.call_history: List[Dict[str, Any]] = []
        self._setup_handlers()
    
    def _setup_handlers(self):
        """Configure les handlers MCP pour les outils disponibles"""
        
        @self.server.list_tools()
        async def list_tools() -> List[Tool]:
            return [
                Tool(
                    name="llm_complete",
                    description="Génère une completion LLM (mockée pour tests)",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "model": {"type": "string"},
                            "prompt": {"type": "string"},
                            "max_tokens": {"type": "integer", "default": 1000},
                            "temperature": {"type": "number", "default": 0.7}
                        },
                        "required": ["model", "prompt"]
                    }
                ),
                Tool(
                    name="llm_chat",
                    description="Génère une réponse de chat (mockée pour tests)",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "model": {"type": "string"},
                            "messages": {"type": "array"},
                            "temperature": {"type": "number", "default": 0.7}
                        },
                        "required": ["model", "messages"]
                    }
                )
            ]
        
        @self.server.call_tool()
        async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
            # Enregistrer l'appel pour analyse
            self.call_history.append({
                "tool": name,
                "args": arguments,
                "timestamp": asyncio.get_event_loop().time()
            })
            
            if name == "llm_complete":
                return await self._handle_complete(arguments)
            elif name == "llm_chat":
                return await self._handle_chat(arguments)
            else:
                raise ValueError(f"Outil inconnu: {name}")
    
    async def _handle_complete(self, args: Dict[str, Any]) -> List[TextContent]:
        """Handle completions mockées"""
        model = args.get("model", "gpt-4.1")
        prompt = args.get("prompt", "")
        
        # Vérifier si on a une réponse préconfigurée
        cache_key = f"complete:{model}:{hash(prompt) % 1000}"
        if cache_key in self.responses:
            mock = self.responses[cache_key]
            return [TextContent(type="text", text=mock.content)]
        
        # Réponse par défaut pour tests
        default_response = f"[MOCK] Réponse pour '{prompt[:50]}...' via {model}"
        return [TextContent(type="text", text=default_response)]
    
    async def _handle_chat(self, args: Dict[str, Any]) -> List[TextContent]:
        """Handle chat completions mockées"""
        model = args.get("model", "gpt-4.1")
        messages = args.get("messages", [])
        
        last_message = messages[-1]["content"] if messages else ""
        
        # Réponse mockée structurée
        response = {
            "model": model,
            "choices": [{
                "message": {
                    "role": "assistant",
                    "content": f"[MOCK CHAT] Réponse à: {last_message[:100]}"
                },
                "finish_reason": "stop"
            }]
        }
        
        return [TextContent(type="text", text=json.dumps(response))]
    
    def register_response(self, model: str, prompt_pattern: str, response: str):
        """Enregistre une réponse mockée pour un pattern donné"""
        cache_key = f"complete:{model}:{hash(prompt_pattern) % 1000}"
        self.responses[cache_key] = MockLLMResponse(
            model=model,
            content=response
        )
    
    def get_call_history(self) -> List[Dict[str, Any]]:
        """Retourne l'historique des appels pour assertions"""
        return self.call_history.copy()
    
    def reset_history(self):
        """Efface l'historique des appels"""
        self.call_history.clear()


Instance globale pour les tests
_mock_server = MCPMockServer()

def get_mock_server() -> MCPMockServer:
    return _mock_server

3. Tests unitaires avec Pytest et Mocking

"""
test_mcp_llm_mocking.py
Tests unitaires pour le framework MCP Tool avec mocking LLM
"""

import pytest
import asyncio
import json
from unittest.mock import patch, MagicMock, AsyncMock
from mock_mcp_server import MCPMockServer, get_mock_server

Configuration pour HolySheep AI
HOLYSHEEP_CONFIG = {
    "base_url": "https://api.holysheep.ai/v1",
    "api_key": "YOUR_HOLYSHEEP_API_KEY",  # Remplacer par votre clé de test
    "default_model": "gpt-4.1"
}

class TestMCPMockServer:
    """Tests pour le serveur mock MCP"""
    
    @pytest.fixture
    def mock_server(self):
        """Fixture qui reset le serveur mock avant chaque test"""
        server = get_mock_server()
        server.reset_history()
        server.responses.clear()
        return server
    
    @pytest.mark.asyncio
    async def test_llm_complete_mock(self, mock_server):
        """Test basique d'une completion mockée"""
        # Enregistrer une réponse
        mock_server.register_response(
            model="gpt-4.1",
            prompt_pattern="Bonjour",
            response="Bonjour ! Comment puis-je vous aider ?"
        )
        
        # Appeler le handler
        result = await mock_server._handle_complete({
            "model": "gpt-4.1",
            "prompt": "Bonjour monde"
        })
        
        assert len(result) == 1
        assert "MOCK" in result[0].text
        print(f"✅ Completion mockée réussie: {result[0].text[:50]}")
    
    @pytest.mark.asyncio
    async def test_llm_chat_mock(self, mock_server):
        """Test du endpoint chat"""
        result = await mock_server._handle_chat({
            "model": "claude-sonnet-4.5",
            "messages": [
                {"role": "user", "content": "Explique les mocks LLM"}
            ]
        })
        
        assert len(result) == 1
        response_data = json.loads(result[0].text)
        assert "choices" in response_data
        assert response_data["model"] == "claude-sonnet-4.5"
        print(f"✅ Chat mocké réussi: modèle={response_data['model']}")
    
    @pytest.mark.asyncio
    async def test_call_history_tracking(self, mock_server):
        """Vérifie que l'historique des appels est bien maintenu"""
        await mock_server._handle_complete({
            "model": "deepseek-v3.2",
            "prompt": "Test prompt"
        })
        
        history = mock_server.get_call_history()
        assert len(history) == 1
        assert history[0]["tool"] == "llm_complete"
        assert history[0]["args"]["model"] == "deepseek-v3.2"
        print(f"✅ Historique追踪: {len(history)} appel(s) enregistré(s)")


class TestLLMIntegrationWithHolySheep:
    """Tests d'intégration avec l'API HolySheep (mode réelles pour validation)"""
    
    @pytest.fixture
    def openai_client(self):
        """Client configuré pour HolySheep AI"""
        from openai import AsyncOpenAI
        
        return AsyncOpenAI(
            base_url=HOLYSHEEP_CONFIG["base_url"],
            api_key=HOLYSHEEP_CONFIG["api_key"],
            timeout=30.0,
            max_retries=3
        )
    
    @pytest.mark.asyncio
    @pytest.mark.integration
    async def test_holy_sheep_connection(self, openai_client):
        """Test de connexion réelle à HolySheep (skip si pas de clé valide)"""
        try:
            response = await openai_client.chat.completions.create(
                model="gpt-4.1",
                messages=[{"role": "user", "content": "Réponds simplement: OK"}],
                max_tokens=10
            )
            
            assert response.choices[0].message.content == "OK"
            assert response.model == "gpt-4.1"
            
            # Vérifier les métadonnées de latence
            usage = response.usage
            print(f"✅ HolySheep connecté: latence={response.id}, tokens={usage.total_tokens}")
            
        except Exception as e:
            pytest.skip(f"Pas de connexion HolySheep: {e}")
    
    @pytest.mark.asyncio
    async def test_mock_vs_real_comparison(self, mock_server, openai_client):
        """Compare le comportement mock vs réel pour validation"""
        test_prompt = "Qu'est-ce que 2+2 ?"
        
        # 1. Test avec mock
        mock_result = await mock_server._handle_complete({
            "model": "gpt-4.1",
            "prompt": test_prompt
        })
        mock_response = mock_result[0].text
        
        # 2. Test avec HolySheep réel (optionnel)
        try:
            real_response = await openai_client.chat.completions.create(
                model="gpt-4.1",
                messages=[{"role": "user", "content": test_prompt}],
                max_tokens=50
            )
            real_content = real_response.choices[0].message.content
            
            print(f"📊 Comparaison:")
            print(f"   Mock:   {mock_response[:60]}")
            print(f"   HolySheep: {real_content[:60]}")
            
            # Les deux doivent contenir quelque chose
            assert len(mock_response) > 0
            assert len(real_content) > 0
            
        except Exception:
            print("⚠️ Test réel ignoré (pas de connexion)")


class TestMCPToolWithUnittest:
    """Tests avec unittest.mock pour les scénarios complexes"""
    
    def test_async_mock_with_patch(self):
        """Utilise unittest.mock pour mocker les appels async"""
        async def fake_llm_call(*args, **kwargs):
            return {"choices": [{"message": {"content": "Mocked response"}}]}
        
        with patch('openai.AsyncOpenAI.chat.completions.create', new_callable=AsyncMock) as mock_create:
            mock_create.return_value = fake_llm_call()
            
            # Votre code qui utilise le client
            async def call_llm():
                from openai import AsyncOpenAI
                client = AsyncOpenAI(
                    base_url=HOLYSHEEP_CONFIG["base_url"],
                    api_key="test-key"
                )
                return await client.chat.completions.create(
                    model="gpt-4.1",
                    messages=[{"role": "user", "content": "test"}]
                )
            
            result = asyncio.run(call_llm())
            assert "Mocked response" in result["choices"][0]["message"]["content"]
            print("✅ Mock avec patchasync successful")
    
    def test_response_validation(self):
        """Valide que le mock retourne une structure conforme"""
        from responses import ResponsesMock
        import responses
        
        with responses.RequestsMock() as rsps:
            rsps.add(
                rsps.POST,
                f"{HOLYSHEEP_CONFIG['base_url']}/chat/completions",
                json={
                    "id": "mock-123",
                    "model": "gpt-4.1",
                    "choices": [{
                        "message": {"role": "assistant", "content": "Test OK"},
                        "finish_reason": "stop"
                    }],
                    "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}
                },
                status=200
            )
            
            # Simuler l'appel HTTP
            import httpx
            response = httpx.post(
                f"{HOLYSHEEP_CONFIG['base_url']}/chat/completions",
                headers={"Authorization": f"Bearer {HOLYSHEEP_CONFIG['api_key']}"},
                json={
                    "model": "gpt-4.1",
                    "messages": [{"role": "user", "content": "test"}]
                }
            )
            
            assert response.status_code == 200
            data = response.json()
            assert data["choices"][0]["message"]["content"] == "Test OK"
            print(f"✅ Response validation: {data['usage']['total_tokens']} tokens")


class TestErrorHandling:
    """Tests des scénarios d'erreur"""
    
    @pytest.mark.asyncio
    async def test_rate_limit_simulation(self):
        """Simule une erreur de rate limit"""
        from openai import RateLimitError
        
        async def mock_rate_limited(*args, **kwargs):
            raise RateLimitError(
                message="Rate limit exceeded",
                response=MagicMock(status_code=429),
                body=None
            )
        
        with patch('openai.AsyncOpenAI.chat.completions.create', new_callable=AsyncMock) as mock:
            mock.side_effect = mock_rate_limited
            
            from openai import AsyncOpenAI
            client = AsyncOpenAI(
                base_url=HOLYSHEEP_CONFIG["base_url"],
                api_key="test"
            )
            
            with pytest.raises(RateLimitError):
                await client.chat.completions.create(
                    model="gpt-4.1",
                    messages=[{"role": "user", "content": "test"}]
                )
            print("✅ Rate limit correctement géré")
    
    @pytest.mark.asyncio
    async def test_timeout_handling(self):
        """Test le comportement timeout"""
        async def mock_timeout(*args, **kwargs):
            raise asyncio.TimeoutError()
        
        with patch('openai.AsyncOpenAI.chat.completions.create', new_callable=AsyncMock) as mock:
            mock.side_effect = mock_timeout
            
            from openai import AsyncOpenAI
            client = AsyncOpenAI(
                base_url=HOLYSHEEP_CONFIG["base_url"],
                api_key="test",
                timeout=0.001
            )
            
            with pytest.raises(asyncio.TimeoutError):
                await client.chat.completions.create(
                    model="gpt-4.1",
                    messages=[{"role": "user", "content": "test"}]
                )
            print("✅ Timeout correctement géré")


if __name__ == "__main__":
    pytest.main([__file__, "-v", "-s"])

4. Configuration CI/CD pour les tests

# pytest.ini
[pytest]
asyncio_mode = auto
markers =
    integration: tests requiring real API connection
    unit: tests with mocks only
testpaths = tests
addopts = -v --tb=short --strict-markers

.env.test pour les environnements CI
HolySheep AI - Credits de test
HOLYSHEEP_API_KEY
Ressources connexes
📚 Tutoriels API IA
💰 Voir les tarifs
📖 Documentation
🚀 Inscription gratuite
Articles connexes
AI API Multi-Cloud Disaster Recovery : Architecture Trois-Zo
Agent 安全护栏 : Les 5 Couches de Défense Contre les Injections 
多 Agent 资源竞争：分布式锁与任务队列协调机制