Go to Qiita Advent Calendar Top
LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@555hamano(Akr)

Model Context Protocol完全解説 30日間シリーズ - Day 28【MCP実戦 #28】既存システム改造：レガシーツールをMCP対応にするアプローチ

Posted at

はじめに

この記事は、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サーバーは最小限の権限で実行
  • ネットワーク分離: レガシーシステムとの通信は専用ネットワーク内で実施
  • 認証情報の管理: 環境変数や専用の設定ファイルで認証情報を管理

エラーハンドリング

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

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

  1. 非侵襲的アプローチ: レガシーシステムを直接改造するのではなく、MCPサーバー内に専用のラッパー（Adapter）を実装する

  2. 適切なパターンの選択:

    • CLI提供 → 外部コマンド実行ラッパー
    • ライブラリ/SDK提供 → ライブラリラッパー
    • HTTP API提供 → HTTP APIラッパー

  3. 堅牢性の確保: エラーハンドリング、タイムアウト、入力検証を徹底的に実装

  4. 観測可能性: 適切なログ出力とモニタリングで運用時の問題を早期発見

  5. 段階的統合: 重要度の低い機能から始めて、徐々に統合範囲を拡大

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

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

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?