AIアプリケーションのユニットテストは、従来のソフトウェアテストとは根本的に異なります。外部APIへの依存、网络延迟、応答の非決定性 — これらすべてがテストの安定性を脅かします。私は以前、MCP（Model Context Protocol）ベースのツールをテストする際、何度も ConnectionError: timeout や 401 Unauthorized に遭遇し、CI/CDパイプラインが不安定になる問題に頭を悩ませました。
本記事では、HolySheep AI を活用したMCP Toolのユニットテスト戦略と、LLMコールの効果的なモック技法について詳しく解説します。
MCP Tool テストの根本課題
MCPプロトコルを使ってAI服务と連携するツールでは、以下の課題に直面します：
- ネットワーク依存性: 実際のAPIコールは遅延・タイムアウト・エラーを起こす
- コスト発生: 本番环境的测试会导致实际的API使用料
- 応答の不確定性: LLMの出力は температура により変動する
- レート制限: テスト実行中にAPIのレートリミットに到達するリスク
基本的なモック戦略：unittest.mock の活用
最もシンプルな方法は、Pythonの unittest.mock を使用してAPIコールをスタブ化することです。以下の例では、HolySheep AIのAPIに対する呼び出しをモックしています：
import unittest
from unittest.mock import patch, MagicMock
from mcp_tools.example_tool import MCPFinancialAnalyzer
import json
class TestMCPFinancialAnalyzer(unittest.TestCase):
"""HolySheep AI APIをモックしたユニットテスト"""
def setUp(self):
"""テスト前の共通設定"""
self.mock_response = {
"id": "mock-chatcmpl-123",
"object": "chat.completion",
"model": "gpt-4o-mini",
"choices": [{
"message": {
"role": "assistant",
"content": json.dumps({
"summary": "第3四半期の売上は前年比15%増加",
"key_metrics": {
"revenue": "1.2億円",
"growth_rate": "15%"
}
})
},
"finish_reason": "stop"
}]
}
@patch('mcp_tools.example_tool.requests.post')
def test_financial_analysis_success(self, mock_post):
"""正常系のテストケース"""
# モック設定： HolySheep AI APIの応答をスタブ化
mock_post.return_value = MagicMock(
status_code=200,
json=lambda: self.mock_response,
headers={'x-ratelimit-remaining': '999'}
)
analyzer = MCPFinancialAnalyzer(api_key="test-key")
result = analyzer.analyze("第3四半期の決算データ")
self.assertIn("summary", result)
self.assertEqual(result["growth_rate"], "15%")
mock_post.assert_called_once()
# APIエンドポイントが正しいことを確認
call_args = mock_post.call_args
self.assertEqual(
call_args[0][0],
"https://api.holysheep.ai/v1/chat/completions"
)
@patch('mcp_tools.example_tool.requests.post')
def test_api_timeout_handling(self, mock_post):
"""タイムアウトエラーのテスト"""
import requests
mock_post.side_effect = requests.exceptions.Timeout(
"Connection timed out after 30 seconds"
)
analyzer = MCPFinancialAnalyzer(api_key="test-key")
with self.assertRaises(requests.exceptions.Timeout) as context:
analyzer.analyze("任意のテキスト")
self.assertIn("timed out", str(context.exception))
@patch('mcp_tools.example_tool.requests.post')
def test_authentication_failure(self, mock_post):
"""認証エラー（401 Unauthorized）のテスト"""
mock_post.return_value = MagicMock(
status_code=401,
json=lambda: {
"error": {
"message": "Incorrect API key provided",
"type": "invalid_request_error"
}
}
)
analyzer = MCPFinancialAnalyzer(api_key="invalid-key")
with self.assertRaises(Exception) as context:
analyzer.analyze("テストデータ")
self.assertIn("401", str(context.exception))
if __name__ == '__main__':
unittest.main(verbosity=2)
高度なモック技法：response_fixture の活用
実際のプロジェクトでは、複数のテストケースで異なる応答を返す必要があります。Pytestの fixture を使用して、柔軟で再利用可能なモック環境構築します：
# conftest.py - テストフィクスチャの定義
import pytest
import json
import httpx
from unittest.mock import AsyncMock
from typing import Dict, Any, List
HolySheep AI の実際の出力形式を模倣
def create_mock_response(
content: str,
model: str = "gpt-4o-mini",
tokens_used: int = 150
) -> Dict[str, Any]:
"""モック応答を生成するヘルパー関数"""
return {
"id": f"mock-{hash(content) % 10000}",
"object": "chat.completion",
"created": 1700000000,
"model": model,
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": content
},
"finish_reason": "stop"
}],
"usage": {
"prompt_tokens": tokens_used // 3,
"completion_tokens": tokens_used // 3,
"total_tokens": tokens_used
},
"x-ratelimit-remaining": "999"
}
class MockHolySheepAPI:
"""HolySheep AI APIの完全なモック実装"""
def __init__(self, responses: List[Dict] = None):
self.responses = responses or []
self.call_history = []
self.current_index = 0
def add_response(self, response: Dict):
"""テスト用の応答を追加"""
self.responses.append(response)
def post(self, url: str, headers: Dict, json: Dict, timeout: int) -> 'MockResponse':
"""HTTP POSTリクエストを模倣"""
self.call_history.append({
"url": url,
"headers": headers,
"json": json,
"timeout": timeout
})
if self.current_index < len(self.responses):
response = self.responses[self.current_index]
self.current_index += 1
return MockResponse(response, 200)
return MockResponse({}, 500)
def reset(self):
"""呼び出し履歴をリセット"""
self.call_history = []
self.current_index = 0
class MockResponse:
"""httpx.Response のモック"""
def __init__(self, data: Dict, status: int):
self._data = data
self.status_code = status
def json(self):
return self._data
@pytest.fixture
def mock_holysheep():
"""共通モックフィクスチャ"""
mock = MockHolySheepAPI()
mock.add_response(
create_mock_response(
json.dumps({
"result": "分析完了",
"confidence": 0.95,
"data": [1, 2, 3, 4, 5]
})
)
)
return mock
@pytest.fixture
def sample_mcp_config():
"""MCP Tool設定のフィクスチャ"""
return {
"api_base": "https://api.holysheep.ai/v1",
"api_key": "test-key-12345",
"model": "gpt-4o-mini",
"max_tokens": 1000,
"temperature": 0.7,
"timeout": 30
}
# test_mcp_tools.py - MCP Toolの本格的なユニットテスト
import pytest
from unittest.mock import patch, MagicMock
from mcp_tools import MCPDataProcessor, MCPQueryEngine
class TestMCPDataProcessor:
"""MCP Data Processor のユニットテストスイート"""
def test_process_structured_data(self, mock_holysheep, sample_mcp_config):
"""構造化データ処理のテスト"""
with patch('mcp_tools.MCPDataProcessor._call_api') as mock_call:
mock_call.return_value = {
"structured_output": {
"items": [
{"id": 1, "value": "test1"},
{"id": 2, "value": "test2"}
],
"count": 2
}
}
processor = MCPDataProcessor(config=sample_mcp_config)
result = processor.process([
{"raw": "データ1"},
{"raw": "データ2"}
])
assert result["structured_output"]["count"] == 2
assert len(result["structured_output"]["items"]) == 2
def test_batch_processing_cost_optimization(self, sample_mcp_config):
"""バッチ処理でのコスト最適化のテスト"""
# HolySheep AI は ¥1=$1（公式¥7.3比85%節約）
# 100回のバッチ処理を想定したコスト計算
mock_responses = []
for i in range(100):
mock_responses.append({
"choices": [{
"message": {
"content": f"Processed item {i}"
}
}],
"usage": {
"total_tokens": 50
}
})
with patch('mcp_tools.MCPDataProcessor._call_api') as mock_call:
mock_call.side_effect = mock_responses
processor = MCPDataProcessor(config=sample_mcp_config)
results = processor.batch_process([f"item_{i}" for i in range(100)])
total_tokens = sum(r["usage"]["total_tokens"] for r in mock_responses)
# HolySheep AI: GPT-4o-mini は $0.15/MTok（推定）
estimated_cost = (total_tokens / 1_000_000) * 0.15
assert estimated_cost < 0.01 # 100件で$0.01未満
@pytest.mark.asyncio
async def test_async_streaming_response(self, sample_mcp_config):
"""非同期ストリーミング応答のテスト"""
async def mock_stream_response():
chunks = [
b'data: {"content": "Hello"}\n\n',
b'data: {"content": " World"}\n\n',
b'data: [DONE]\n\n'
]
for chunk in chunks:
yield chunk
with patch('httpx.AsyncClient.stream') as mock_stream:
mock_stream.return_value.__aenter__.return_value.aiter_lines = mock_stream_response
processor = MCPQueryEngine(config=sample_mcp_config)
result = await processor.stream_query("テストクエリ")
assert "Hello" in result
assert "World" in result
class TestMCPErrorHandling:
"""エラーハンドリングのテスト"""
def test_rate_limit_retry_logic(self, sample_mcp_config):
"""レート制限時のリトライロジック"""
import time
call_count = [0]
def mock_call_with_rate_limit(*args, **kwargs):
call_count[0] += 1
if call_count[0] < 3:
return MagicMock(
status_code=429,
headers={'retry-after': '1'},
json=lambda: {"error": "Rate limit exceeded"}
)
return MagicMock(
status_code=200,
json=lambda: {"result": "success"}
)
with patch('mcp_tools.MCPDataProcessor._call_api', side_effect=mock_call_with_rate_limit):
processor = MCPDataProcessor(config=sample_mcp_config)
result = processor.process_with_retry("test", max_retries=5)
assert call_count[0] == 3
assert result["result"] == "success"
def test_context_length_exceeded(self, sample_mcp_config):
"""コンテキスト長超過エラーのテスト"""
with patch('mcp_tools.MCPDataProcessor._call_api') as mock_call:
mock_call.return_value = MagicMock(
status_code=400,
json=lambda: {
"error": {
"message": "This model's maximum context length is 128000 tokens",
"type": "invalid_request_error",
"param": "messages",
"code": "context_length_exceeded"
}
}
)
processor = MCPDataProcessor(config=sample_mcp_config)
with pytest.raises(Exception) as exc_info:
processor.process("x" * 200000) # 200Kトークン超
assert "context_length" in str(exc_info.value).lower()
統合テスト：実際のAPIを活用したテスト戦略
モックテストに加えて、実際にAPIを呼び出す統合テストも重要です。HolySheep AIの <50ms という低遅延特性を利用すれば、テスト環境でも迅速に検証できます：
# integration_test.py - 統合テスト（実際のAPIコール）
import pytest
import httpx
import time
from typing import Dict, Any
class TestHolySheepAIIntegration:
"""HolySheep AI APIとの実際の統合テスト"""
@pytest.fixture(autouse=True)
def setup(self):
"""テスト前の共通設定"""
self.api_key = "YOUR_HOLYSHEEP_API_KEY" # 環境変数から取得推奨
self.base_url = "https://api.holysheep.ai/v1"
self.client = httpx.Client(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=30.0
)
yield
self.client.close()
def test_api_latency_performance(self):
"""API応答遅延の性能テスト"""
latencies = []
for i in range(10):
start = time.time()
response = self.client.post("/chat/completions", json={
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": f"Reply with: {i}"}],
"max_tokens": 10
})
elapsed = (time.time() - start) * 1000 # ミリ秒に変換
assert response.status_code == 200
latencies.append(elapsed)
avg_latency = sum(latencies) / len(latencies)
p95_latency = sorted(latencies)[int(len(latencies) * 0.95)]
# HolySheep AIは <50ms レイテンシを保証
print(f"平均レイテンシ: {avg_latency:.2f}ms")
print(f"P95レイテンシ: {p95_latency:.2f}ms")
assert p95_latency < 200 # P95で200ms以内
def test_cost_calculation(self):
"""コスト計算の正確性テスト"""
response = self.client.post("/chat/completions", json={
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": "Explain quantum computing in 50 words"}],
"max_tokens": 100
})
assert response.status_code == 200
data = response.json()
usage = data.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", 0)
# トークン数の検証
assert total_tokens == prompt_tokens + completion_tokens
assert total_tokens > 0
# コスト計算（HolySheep AI: GPT-4o-mini $0.15/MTok入力、$0.60/MTok出力）
input_cost = (prompt_tokens / 1_000_000) * 0.15
output_cost = (completion_tokens / 1_000_000) * 0.60
print(f"入力トークン: {prompt_tokens}, 出力トークン: {completion_tokens}")
print(f"推定コスト: ${input_cost + output_cost:.6f}")
def test_model_availability(self):
"""利用可能なモデルの一覧取得テスト"""
# HolySheep AI 利用可能モデル:
# - GPT-4.1: $8/MTok
# - Claude Sonnet 4.5: $15/MTok
# - Gemini 2.5 Flash: $2.50/MTok
# - DeepSeek V3.2: $0.42/MTok
models_to_test = [
"gpt-4o-mini",
"claude-sonnet-4-20250514",
"gemini-2.0-flash",
"deepseek-v3.2"
]
for model in models_to_test:
response = self.client.post("/chat/completions", json={
"model": model,
"messages": [{"role": "user", "content": "Hi"}],
"max_tokens": 5
})
# モデルが利用不可の場合は400が返る可能性あり
assert response.status_code in [200, 400]
print(f"{model}: {response.status_code}")
CI/CDパイプラインへの統合
GitHub Actionsを活用した自動テストパイプラインの構築：
# .github/workflows/mcp-test.yml
name: MCP Tool Unit Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
env:
HOLYSHEEP_API_KEY: ${{ secrets.HOLYSHEEP_API_KEY }}
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install pytest pytest-asyncio httpx
pip install -e .
- name: Run unit tests (mocked)
run: |
pytest tests/unit/ -v --cov=mcp_tools \
--cov-report=xml \
--cov-fail-under=80
- name: Run integration tests
if: env.HOLYSHEEP_API_KEY != ''
run: |
pytest tests/integration/ -v \
--tb=short \
-k "not slow"
continue-on-error: true
- name: Upload coverage
uses: codecov/codecov-action@v4
with:
file: ./coverage.xml
よくあるエラーと対処法
1. ConnectionError: timeout after 30 seconds
原因: APIエンドポイントへの接続がタイムアウトしました。ネットワーク問題またはエンドポイント不通が考えられます。
# 解決策: タイムアウト設定の最適化とリトライ機構の実装
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_api_with_retry(client: httpx.Client, payload: Dict) -> Dict:
"""指数関数的バックオフでリトライ"""
try:
response = client.post(
"/chat/completions",
json=payload,
timeout=httpx.Timeout(60.0, connect=10.0) # 接続:10s, 読取:60s
)
response.raise_for_status()
return response.json()
except httpx.TimeoutException as e:
print(f"タイムアウト発生: {e}")
raise
except httpx.ConnectError as e:
# エンドポイント確認
print(f"接続エラー: エンドポイント確認必要 - {e}")
raise
2. 401 Unauthorized - Incorrect API key provided
原因: APIキーが無効または期限切れです。キーが正しく環境変数に設定されていない可能性があります。
# 解決策: APIキー検証ロジックの実装
import os
from functools import wraps
def validate_api_key(func):
"""APIキー検証デコレータ"""
@wraps(func)
def wrapper(*args, **kwargs):
api_key = os.environ.get('HOLYSHEEP_API_KEY') or kwargs.get('api_key')
if not api_key:
raise ValueError(
"HOLYSHEEP_API_KEYが設定されていません。"
"https://www.holysheep.ai/register でAPIキーを取得してください。"
)
if api_key == 'YOUR_HOLYSHEEP_API_KEY':
raise ValueError(
"デモ用APIキーを実際のキーに置き換えてください"
)
if len(api_key) < 20:
raise ValueError(f"APIキーが短すぎます（長さ: {len(api_key)}）")
return func(*args, **kwargs)
return wrapper
@validate_api_key
def create_client(api_key: str = None):
"""バリデーション付きクライアント生成"""
return httpx.Client(
base_url="https://api.holysheep.ai/v1",
headers={"Authorization": f"Bearer {api_key}"}
)