ProductMaster MCP Server

#!/usr/bin/env python3 """ ProductMaster MCP Server テストスクリプト """ import asyncio import json import sys import os from typing import Dict, Any # サーバーパスを追加 sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'server')) async def test_mcp_tools(): """MCPツール機能テスト""" print("=== ProductMaster MCP Server テスト ===") try: from mcp_server import ProductMasterMCPServer async with ProductMasterMCPServer() as server: print("\n✅ MCPサーバー初期化成功") # 1. 柔軟検索テスト print("\n1. 柔軟検索テスト:") search_result = await server.search_products_flexible( product_types=["bond"], currencies=["USD"], sort_by="investment_desc", limit=3 ) if search_result.get("success"): print(f" 検索結果: {search_result['total_found']}件見つかりました") for product in search_result['products']: print(f" - {product['product_name']} ({product['currency']}, 投資額: {product.get('minimum_investment', 0)})") else: print(f" エラー: {search_result.get('error')}") # 2. 商品詳細テスト print("\n2. 商品詳細テスト:") details_result = await server.get_product_details("AAPL") if details_result.get("success"): product = details_result['product'] print(f" 商品名: {product['product_name']}") print(f" 発行者: {product.get('issuer', '')}") print(f" 通貨: {product.get('currency', '')}") print(f" 投資額: {product.get('minimum_investment', 0)}") else: print(f" エラー: {details_result.get('error')}") # 3. 統計情報テスト print("\n3. 統計情報テスト:") stats_result = await server.get_product_statistics(group_by="currency") if stats_result.get("success"): print(f" 総商品数: {stats_result['total_products']}") print(f" 通貨別: {stats_result['by_currency']}") print(f" 投資額統計: {stats_result['investment_stats']}") else: print(f" エラー: {stats_result.get('error')}") print("\n✅ 全てのテストが完了しました！") except ImportError as e: print(f"❌ インポートエラー: {e}") print("MCPライブラリがインストールされていない可能性があります") print("pip install mcp httpx pydantic を実行してください") except Exception as e: print(f"❌ テストエラー: {e}") import traceback traceback.print_exc() def test_mcp_protocol(): """MCPプロトコルテスト（stdio transport）""" print("\n=== MCPプロトコルテスト ===") # JSON-RPC 2.0リクエストのテスト test_requests = [ { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "test-client", "version": "1.0.0"} } }, { "jsonrpc": "2.0", "id": 2, "method": "tools/list" } ] print("テスト用JSON-RPCリクエスト:") for i, request in enumerate(test_requests, 1): print(f"{i}. {json.dumps(request, indent=2)}") print("\n手動テスト方法:") print("echo '{\"jsonrpc\": \"2.0\", \"id\": 1, \"method\": \"tools/list\"}' | python3 ../server/mcp_server.py") async def main(): """メインテスト関数""" await test_mcp_tools() test_mcp_protocol() if __name__ == "__main__": asyncio.run(main())