はじめに

この記事は、QiitaのModel Context Protocol（以下、MCP）解説シリーズの第28回です。

今回は、MCPをより大規模なプロジェクトに適用するための実践的なテーマ、既存のレガシーツールをMCP対応にするアプローチについて解説します。これにより、長年使われてきた重要なシステムを、LLMと連携して活用できるようになります。

💡 なぜレガシーシステムをMCP対応にするのか？

多くの企業には、ビジネスの根幹を支えるレガシーシステムが存在します。これらは最新の技術スタックではないものの、独自の専門的な機能や膨大なデータを蓄積している貴重な資産です。

MCPを使ってこれらのシステムをLLMと連携させることで、以下のようなメリットが得られます：

自動化と効率化: LLMがレガシーツールの操作を代行し、手動で行っていた作業を自動化
ナレッジの民主化: 専門家しか使えなかったツールの知識を、LLMを通じて組織全体で活用
投資の保護: 大規模なリファクタリングや刷新を行うことなく、既存の資産を最新のAI技術と組み合わせ
段階的な移行: 完全なシステム刷新の前に、AIとの連携で既存システムの価値を最大化

🛠️ レガシーシステムをMCP対応にするアプローチ

レガシーシステムをMCP対応にするための基本的なアプローチは、**「直接改造しない」ことです。代わりに、「ラッパー（Wrapper）」**を開発し、MCPサーバーを介してレガシーシステムにアクセスさせます。

このアプローチには、主に以下の3つのパターンがあります。

1. 外部コマンド実行ラッパー

レガシーツールがコマンドラインインターフェース（CLI）を提供している場合に有効なアプローチです。MCPサーバーが特定の引数で外部コマンドを実行するToolsを提供します。

実装例：Pythonスクリプトを呼び出す

legacy_tool.pyという、製品情報を返す古いPythonスクリプトがあるとします。

legacy_tool.py

#!/usr/bin/env python3
import sys
import json
import logging

# ログ設定（実際のレガシーシステムではログ出力が重要）
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def get_product_info(product_id):
    """製品情報を取得する関数
    
    実際のレガシーシステムでは、データベース接続や
    複雑なビジネスロジックが含まれることが多い
    """
    logger.info(f"製品情報を取得中: {product_id}")
    
    # 実際はデータベースから情報を取得するなどの処理
    products = {
        "P001": {"name": "Laptop", "price": 1200, "category": "Electronics"},
        "P002": {"name": "Monitor", "price": 300, "category": "Electronics"},
        "P003": {"name": "Keyboard", "price": 80, "category": "Accessories"},
    }
    return products.get(product_id, None)

def main():
    if len(sys.argv) < 2:
        print(json.dumps({"error": "製品IDが指定されていません"}), file=sys.stderr)
        sys.exit(1)
    
    product_id = sys.argv[1]
    
    try:
        info = get_product_info(product_id)
        if info:
            result = {"success": True, "data": info}
            print(json.dumps(result, ensure_ascii=False))
        else:
            result = {"success": False, "error": "製品が見つかりません"}
            print(json.dumps(result, ensure_ascii=False))
            sys.exit(1)
    except Exception as e:
        logger.error(f"エラーが発生しました: {str(e)}")
        result = {"success": False, "error": str(e)}
        print(json.dumps(result, ensure_ascii=False), file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()

このスクリプトを呼び出すMCPサーバーのToolsを定義します。

mcp_server.py

import subprocess
import json
import logging
from typing import Dict, Any
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool,
    TextContent,
    CallToolRequest,
    CallToolResult,
)
from pydantic import BaseModel, Field

# ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# MCPサーバーのインスタンス作成
server = Server("legacy-product-server")

class GetProductInfoInput(BaseModel):
    product_id: str = Field(..., description="取得したい製品のID（例: P001, P002, P003）")

@server.call_tool()
async def get_product_info(arguments: Dict[str, Any]) -> list[TextContent]:
    """レガシーシステムから製品情報を取得します"""
    try:
        # 引数の検証
        input_data = GetProductInfoInput(**arguments)
        
        logger.info(f"製品情報取得開始: {input_data.product_id}")
        
        # subprocessを使って外部コマンドを実行
        result = subprocess.run(
            ['python3', 'legacy_tool.py', input_data.product_id],
            capture_output=True,
            text=True,
            timeout=30,  # タイムアウトを設定
            check=False  # エラーコードをチェックしない（手動で処理）
        )
        
        if result.returncode == 0:
            # 成功時の処理
            response_data = json.loads(result.stdout)
            logger.info(f"製品情報取得成功: {input_data.product_id}")
            return [TextContent(
                type="text",
                text=json.dumps(response_data, ensure_ascii=False, indent=2)
            )]
        else:
            # エラー時の処理
            error_message = result.stderr or result.stdout
            logger.error(f"外部コマンド実行エラー: {error_message}")
            return [TextContent(
                type="text", 
                text=json.dumps({
                    "success": False,
                    "error": f"レガシーシステムエラー: {error_message}"
                }, ensure_ascii=False, indent=2)
            )]
            
    except subprocess.TimeoutExpired:
        logger.error("外部コマンドがタイムアウトしました")
        return [TextContent(
            type="text",
            text=json.dumps({
                "success": False,
                "error": "処理がタイムアウトしました"
            }, ensure_ascii=False, indent=2)
        )]
    except json.JSONDecodeError as e:
        logger.error(f"JSON解析エラー: {str(e)}")
        return [TextContent(
            type="text",
            text=json.dumps({
                "success": False,
                "error": "レガシーシステムからの応答が不正です"
            }, ensure_ascii=False, indent=2)
        )]
    except Exception as e:
        logger.error(f"予期しないエラー: {str(e)}")
        return [TextContent(
            type="text",
            text=json.dumps({
                "success": False,
                "error": f"予期しないエラー: {str(e)}"
            }, ensure_ascii=False, indent=2)
        )]

@server.list_tools()
async def list_tools() -> list[Tool]:
    """利用可能なツールのリストを返す"""
    return [
        Tool(
            name="get_product_info",
            description="レガシー製品管理システムから製品情報を取得します。製品ID（P001, P002, P003など）を指定してください。",
            inputSchema={
                "type": "object",
                "properties": {
                    "product_id": {
                        "type": "string",
                        "description": "取得したい製品のID（例: P001, P002, P003）"
                    }
                },
                "required": ["product_id"]
            }
        )
    ]

async def main():
    """MCPサーバーのメイン関数"""
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="legacy-product-server",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=None,
                    experimental_capabilities={}
                )
            )
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

2. ライブラリ/SDKラッパー

レガシーシステムが言語ごとのライブラリやSDKを提供している場合に有効です。MCPサーバーがそのライブラリを直接インポートして利用します。

実装例：Pythonライブラリを呼び出す

legacy_lib.pyという古いPythonライブラリがあるとします。

legacy_lib.py

"""
レガシーなデータ処理ライブラリ
実際のレガシーシステムでは、複雑なビジネスロジックが含まれる
"""
import time
import logging
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)

class DataProcessor:
    """古いデータ処理クラス"""
    
    def __init__(self, config: Optional[Dict] = None):
        self.config = config or {}
        logger.info("DataProcessor初期化完了")
    
    def process_data(self, data: str, options: Optional[Dict] = None) -> Dict[str, Any]:
        """データ処理のメイン関数"""
        logger.info(f"データ処理開始: {len(data)}文字")
        
        # 実際のレガシーシステムでは時間のかかる処理が含まれることが多い
        time.sleep(0.1)  # シミュレート
        
        processed_options = options or {}
        case_transform = processed_options.get('case', 'upper')
        
        if case_transform == 'upper':
            processed_text = data.upper()
        elif case_transform == 'lower':
            processed_text = data.lower()
        else:
            processed_text = data
        
        result = {
            "original_length": len(data),
            "processed_data": processed_text,
            "processing_time": 0.1,
            "options_used": processed_options
        }
        
        logger.info("データ処理完了")
        return result
    
    def validate_data(self, data: str) -> bool:
        """データの妥当性をチェック"""
        return len(data) > 0 and len(data) < 10000

# グローバル関数（後方互換性のため）
def process_data(data: str) -> Dict[str, Any]:
    """シンプルなデータ処理関数（レガシー互換）"""
    processor = DataProcessor()
    return processor.process_data(data)

このライブラリを呼び出すMCPサーバーを実装します。

mcp_legacy_server.py

import logging
from typing import Dict, Any
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool,
    TextContent,
    CallToolRequest,
    CallToolResult,
)
import json

# レガシーライブラリのインポート
import legacy_lib

# ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# MCPサーバーのインスタンス作成
server = Server("legacy-data-processor")

# レガシーライブラリのインスタンスを作成
data_processor = legacy_lib.DataProcessor({
    "max_length": 10000,
    "timeout": 30
})

@server.call_tool()
async def process_data_legacy(arguments: Dict[str, Any]) -> list[TextContent]:
    """レガシーライブラリを使ってデータを処理します"""
    try:
        # 必須パラメータのチェック
        if 'data' not in arguments:
            return [TextContent(
                type="text",
                text=json.dumps({
                    "success": False,
                    "error": "dataパラメータが必要です"
                }, ensure_ascii=False, indent=2)
            )]
        
        data = str(arguments['data'])
        options = arguments.get('options', {})
        
        # データの妥当性チェック
        if not data_processor.validate_data(data):
            return [TextContent(
                type="text",
                text=json.dumps({
                    "success": False,
                    "error": "データが不正です（空文字または10000文字を超える）"
                }, ensure_ascii=False, indent=2)
            )]
        
        logger.info(f"レガシーデータ処理開始: {len(data)}文字")
        
        # レガシーライブラリを直接呼び出す
        result = data_processor.process_data(data, options)
        
        # 成功時のレスポンス
        response = {
            "success": True,
            "result": result
        }
        
        logger.info("レガシーデータ処理完了")
        return [TextContent(
            type="text",
            text=json.dumps(response, ensure_ascii=False, indent=2)
        )]
        
    except Exception as e:
        logger.error(f"レガシーライブラリ実行エラー: {str(e)}")
        return [TextContent(
            type="text",
            text=json.dumps({
                "success": False,
                "error": f"レガシーライブラリエラー: {str(e)}"
            }, ensure_ascii=False, indent=2)
        )]

@server.call_tool()
async def validate_data_legacy(arguments: Dict[str, Any]) -> list[TextContent]:
    """レガシーシステムでデータの妥当性をチェックします"""
    try:
        if 'data' not in arguments:
            return [TextContent(
                type="text",
                text=json.dumps({
                    "success": False,
                    "error": "dataパラメータが必要です"
                }, ensure_ascii=False, indent=2)
            )]
        
        data = str(arguments['data'])
        is_valid = data_processor.validate_data(data)
        
        return [TextContent(
            type="text",
            text=json.dumps({
                "success": True,
                "is_valid": is_valid,
                "data_length": len(data),
                "validation_rules": "長さが1文字以上10000文字以下"
            }, ensure_ascii=False, indent=2)
        )]
        
    except Exception as e:
        logger.error(f"バリデーションエラー: {str(e)}")
        return [TextContent(
            type="text",
            text=json.dumps({
                "success": False,
                "error": f"バリデーションエラー: {str(e)}"
            }, ensure_ascii=False, indent=2)
        )]

@server.list_tools()
async def list_tools() -> list[Tool]:
    """利用可能なツールのリストを返す"""
    return [
        Tool(
            name="process_data_legacy",
            description="レガシーデータ処理ライブラリを使って文字列データを処理します。大文字・小文字変換などが可能です。",
            inputSchema={
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "処理する文字列データ（1-10000文字）"
                    },
                    "options": {
                        "type": "object",
                        "description": "処理オプション",
                        "properties": {
                            "case": {
                                "type": "string",
                                "enum": ["upper", "lower", "none"],
                                "description": "大文字・小文字変換の指定"
                            }
                        }
                    }
                },
                "required": ["data"]
            }
        ),
        Tool(
            name="validate_data_legacy",
            description="レガシーシステムでデータの妥当性をチェックします。",
            inputSchema={
                "type": "object",
                "properties": {
                    "data": {
                        "type": "string",
                        "description": "妥当性をチェックする文字列データ"
                    }
                },
                "required": ["data"]
            }
        )
    ]

async def main():
    """MCPサーバーのメイン関数"""
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="legacy-data-processor",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=None,
                    experimental_capabilities={}
                )
            )
        )

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

3. HTTP/REST APIラッパー

レガシーシステムがHTTP APIを提供している場合に有効なアプローチです。

http_legacy_wrapper.py

import aiohttp
import json
import logging
from typing import Dict, Any
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

logger = logging.getLogger(__name__)
server = Server("legacy-api-wrapper")

@server.call_tool()
async def call_legacy_api(arguments: Dict[str, Any]) -> list[TextContent]:
    """レガシーHTTP APIを呼び出します"""
    try:
        endpoint = arguments.get('endpoint')
        method = arguments.get('method', 'GET')
        data = arguments.get('data', {})
        
        async with aiohttp.ClientSession() as session:
            if method.upper() == 'GET':
                async with session.get(f"http://legacy-system:8080{endpoint}") as response:
                    result = await response.json()
            else:
                async with session.post(
                    f"http://legacy-system:8080{endpoint}",
                    json=data
                ) as response:
                    result = await response.json()
        
        return [TextContent(
            type="text",
            text=json.dumps({
                "success": True,
                "status_code": response.status,
                "data": result
            }, ensure_ascii=False, indent=2)
        )]
        
    except Exception as e:
        logger.error(f"HTTP API呼び出しエラー: {str(e)}")
        return [TextContent(
            type="text",
            text=json.dumps({
                "success": False,
                "error": str(e)
            }, ensure_ascii=False, indent=2)
        )]

🔧 セキュリティとエラーハンドリングの考慮事項

レガシーシステムとの統合では、以下の点に特に注意が必要です：

セキュリティ対策

入力サニタイゼーション: 外部コマンドに渡すパラメータは必ず検証・サニタイゼーション
権限管理: MCPサーバーは最小限の権限で実行
ネットワーク分離: レガシーシステムとの通信は専用ネットワーク内で実施
認証情報の管理: 環境変数や専用の設定ファイルで認証情報を管理

エラーハンドリング

タイムアウト設定: レガシーシステムの応答が遅い場合を想定
リトライ機能: 一時的な障害に対する自動復旧機能
詳細なログ出力: 問題の特定とデバッグのための包括的なログ

🎯 まとめ：レガシー統合のベストプラクティス

非侵襲的アプローチ: レガシーシステムを直接改造するのではなく、MCPサーバー内に専用のラッパー（Adapter）を実装する
適切なパターンの選択:
- CLI提供 → 外部コマンド実行ラッパー
- ライブラリ/SDK提供 → ライブラリラッパー
- HTTP API提供 → HTTP APIラッパー
堅牢性の確保: エラーハンドリング、タイムアウト、入力検証を徹底的に実装
観測可能性: 適切なログ出力とモニタリングで運用時の問題を早期発見
段階的統合: 重要度の低い機能から始めて、徐々に統合範囲を拡大

このアプローチにより、MCPを企業の既存IT資産と安全かつ効果的に統合できます。レガシーシステムの価値を最大化しながら、将来的なシステム刷新への道筋も描けるでしょう。

次回は、いよいよシリーズ最終章。Day 29では、これまでの知識を総動員した「実践プロジェクト」を解説します。お楽しみに！