f0_make_randomvalues MCP Server

# MCP Server 要件定義・仕様書 ## f0_make_randomvalues Function Block の MCP サーバー実装 ### 1. エグゼクティブサマリー #### 1.1 プロジェクト概要 FGDB（Functional Graph Database）システムで使用される `f0_make_randomvalues` 関数ブロックを、Model Context Protocol (MCP) サーバーとして実装する。これにより、Claude DesktopなどのMCP対応クライアントから、乱数生成機能を直接利用可能にする。 #### 1.2 ビジネス価値 - **相互運用性の向上**: FGDBシステムの機能をLLMアプリケーションに統合 - **開発効率の向上**: 既存の関数ブロックをMCPエコシステムで再利用 - **標準化**: MCP仕様準拠による将来的な拡張性の確保 #### 1.3 主要成果物 - MCPサーバー実装（`mcp_f0_server.py`） - Claude Desktop統合設定 - 自動化テストスイート - 運用ドキュメント --- ## 2. システムアーキテクチャ ### 2.1 全体構成 ``` ┌─────────────────┐ MCP Protocol ┌─────────────────┐ │ MCP Client │◄────────────────────►│ MCP Server │ │ (Claude, etc) │ │ (f0_random) │ └─────────────────┘ └─────────────────┘ │ ▼ ┌─────────────────┐ │ FGDB Function │ │ Block (f0) │ └─────────────────┘ ``` ### 2.2 コンポーネント構成 | コンポーネント | 役割 | 技術スタック | |---------------|------|-------------| | MCPサーバー | プロトコル処理、メッセージ管理 | Python, MCP SDK | | ツールハンドラー | 乱数生成機能の実装 | FastMCP | | リソースプロバイダー | データファイルアクセス | asyncio | | プロンプトテンプレート | ユーザーインタラクション定義 | YAML/JSON | ### 2.3 通信方式 - **プライマリ**: stdio（標準入出力） - **セカンダリ**: Streamable HTTP（将来拡張用） - **プロトコル**: JSON-RPC 2.0 over stdio --- ## 3. 機能要件 ### 3.1 MCPサーバー基本機能 #### 3.1.1 サーバー初期化 ```python @dataclass class ServerCapabilities: tools: bool = True # ツール機能を提供 resources: bool = True # リソース機能を提供 prompts: bool = True # プロンプト機能を提供 logging: bool = True # ロギング機能を有効化 ``` #### 3.1.2 ライフサイクル管理 | フェーズ | 処理内容 | |---------|---------| | 起動 | 環境初期化、デフォルト設定読込 | | 初期化 | クライアント接続、capability交換 | | 実行 | リクエスト処理、レスポンス送信 | | 終了 | リソース解放、ログ保存 | ### 3.2 ツール機能 #### 3.2.1 generate_random_numbers ```python @mcp.tool() def generate_random_numbers( count: int = 5, min_value: int = 0, max_value: int = 999, seed: int | None = None ) -> RandomNumberResult: """Generate random numbers with specified parameters.""" ``` **パラメータ仕様**: | パラメータ | 型 | デフォルト | 説明 | |-----------|-----|-----------|------| | count | int | 5 | 生成する乱数の個数 | | min_value | int | 0 | 乱数の最小値 | | max_value | int | 999 | 乱数の最大値 | | seed | int\|None | None | 乱数シード（再現性用） | **戻り値（構造化出力）**: ```python class RandomNumberResult(BaseModel): numbers: List[int] count: int min_value: int max_value: int timestamp: str seed: Optional[int] = None ``` #### 3.2.2 save_random_data ```python @mcp.tool() def save_random_data( numbers: List[int], filename: str = "data.csv" ) -> SaveResult: """Save random numbers to CSV file.""" ``` #### 3.2.3 analyze_random_data ```python @mcp.tool() def analyze_random_data( numbers: List[int] ) -> StatisticsResult: """Analyze statistical properties of random numbers.""" ``` ### 3.3 リソース機能 #### 3.3.1 動的リソース ```python @mcp.resource("random://current") def get_current_random_data() -> str: """Get the most recently generated random numbers.""" ``` #### 3.3.2 ファイルリソース ```python @mcp.resource("file://data/{filename}") def get_saved_data(filename: str) -> str: """Read saved random data from file.""" ``` #### 3.3.3 メタデータリソース ```python @mcp.resource("meta://statistics") def get_statistics() -> str: """Get statistical metadata about all generated data.""" ``` ### 3.4 プロンプト機能 #### 3.4.1 インタラクティブ生成 ```python @mcp.prompt(title="Generate Random Dataset") def generate_dataset_prompt( purpose: str = "testing", count: int = 10, distribution: str = "uniform" ) -> List[Message]: """Interactive prompt for dataset generation.""" ``` #### 3.4.2 分析レポート生成 ```python @mcp.prompt(title="Analyze Random Data") def analyze_data_prompt( data_source: str ) -> str: """Generate analysis report prompt.""" ``` --- ## 4. 非機能要件 ### 4.1 パフォーマンス要件 | メトリクス | 目標値 | 測定方法 | |-----------|--------|----------| | レスポンスタイム | <100ms | 95パーセンタイル | | 同時接続数 | 10 | 最大同時クライアント | | メモリ使用量 | <200MB | ピーク時使用量 | | CPU使用率 | <20% | アイドル時 | ### 4.2 信頼性要件 - **可用性**: 99.9%（ローカル実行環境） - **エラーハンドリング**: 全例外の適切なキャッチと報告 - **データ整合性**: トランザクショナルな書き込み ### 4.3 セキュリティ要件 - **認証**: OAuth 2.1対応（オプション） - **入力検証**: 全パラメータの型・範囲チェック - **ファイルアクセス**: サンドボックス内のみ ### 4.4 互換性要件 - **MCPバージョン**: 仕様2025-06-18準拠 - **Python**: 3.8以上 - **SDK**: mcp[cli] 最新版 --- ## 5. 実装詳細 ### 5.1 プロジェクト構造 ``` mcp_f0_server/ ├── src/ │ ├── __init__.py │ ├── server.py # MCPサーバーメイン │ ├── tools.py # ツール実装 │ ├── resources.py # リソース実装 │ ├── prompts.py # プロンプト実装 │ └── models.py # Pydanticモデル定義 ├── tests/ │ ├── test_server.py │ ├── test_tools.py │ └── test_integration.py ├── config/ │ └── server_config.yaml ├── pyproject.toml ├── README.md └── requirements.txt ``` ### 5.2 主要実装ファイル #### 5.2.1 server.py - メインサーバー ```python from mcp.server.fastmcp import FastMCP from contextlib import asynccontextmanager from typing import AsyncIterator import logging # ロギング設定 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # アプリケーションコンテキスト @dataclass class AppContext: data_dir: Path current_data: Optional[List[int]] = None @asynccontextmanager async def lifespan(server: FastMCP) -> AsyncIterator[AppContext]: """サーバーライフサイクル管理""" data_dir = Path("./data") data_dir.mkdir(exist_ok=True) context = AppContext(data_dir=data_dir) logger.info("MCP Server initialized") try: yield context finally: logger.info("MCP Server shutting down") # MCPサーバーインスタンス mcp = FastMCP( name="f0_random_server", version="1.0.0", lifespan=lifespan ) ``` #### 5.2.2 models.py - データモデル定義 ```python from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime class RandomNumberResult(BaseModel): """乱数生成結果の構造化データ""" numbers: List[int] = Field(description="生成された乱数リスト") count: int = Field(description="乱数の個数") min_value: int = Field(description="最小値") max_value: int = Field(description="最大値") timestamp: str = Field(description="生成日時") seed: Optional[int] = Field(default=None, description="使用シード") class StatisticsResult(BaseModel): """統計分析結果""" mean: float = Field(description="平均値") median: float = Field(description="中央値") std_dev: float = Field(description="標準偏差") min: int = Field(description="最小値") max: int = Field(description="最大値") count: int = Field(description="データ数") ``` ### 5.3 エラーハンドリング ```python class MCPServerError(Exception): """基底例外クラス""" pass class ValidationError(MCPServerError): """入力検証エラー""" pass class ResourceNotFoundError(MCPServerError): """リソース未検出エラー""" pass @mcp.tool() async def safe_operation(ctx: Context, **kwargs): try: # 処理実装 result = await process_data(**kwargs) return result except ValidationError as e: await ctx.error(f"Validation failed: {e}") raise except Exception as e: await ctx.error(f"Unexpected error: {e}") logger.exception("Operation failed") raise MCPServerError("Internal server error") ``` --- ## 6. テスト戦略 ### 6.1 テストレベル | レベル | カバレッジ目標 | テスト内容 | |--------|--------------|-----------| | ユニットテスト | 90% | 個別関数の動作確認 | | 統合テスト | 80% | MCPプロトコル通信 | | E2Eテスト | 主要シナリオ | Claude Desktop統合 | ### 6.2 テストシナリオ #### 6.2.1 正常系テスト 1. **基本的な乱数生成** - デフォルトパラメータでの生成 - カスタムパラメータでの生成 - シード指定での再現性確認 2. **ファイル操作** - CSV保存と読込 - 複数ファイルの管理 - 同時アクセス処理 3. **リソースアクセス** - 動的リソースの取得 - テンプレートパラメータ解決 #### 6.2.2 異常系テスト 1. **入力検証** - 範囲外パラメータ - 型不一致 - 必須パラメータ欠落 2. **エラーリカバリー** - ファイル書込失敗 - メモリ不足 - タイムアウト ### 6.3 テスト実装例 ```python import pytest from unittest.mock import AsyncMock, patch from mcp.client import ClientSession @pytest.mark.asyncio async def test_generate_random_numbers(): """乱数生成ツールのテスト""" async with create_test_client() as session: result = await session.call_tool( "generate_random_numbers", arguments={"count": 10, "seed": 42} ) assert result.structuredContent is not None data = result.structuredContent assert data["count"] == 10 assert len(data["numbers"]) == 10 assert data["seed"] == 42 @pytest.mark.asyncio async def test_resource_access(): """リソースアクセスのテスト""" async with create_test_client() as session: resource = await session.read_resource( "random://current" ) assert resource.contents content = resource.contents[0] assert isinstance(content, TextContent) ``` --- ## 7. デプロイメント ### 7.1 Claude Desktop統合 #### 7.1.1 設定ファイル（claude_desktop_config.json） ```json { "mcpServers": { "f0_random": { "command": "uv", "args": ["run", "mcp_f0_server.py"], "cwd": "/absolute/path/to/mcp_f0_server", "env": { "LOG_LEVEL": "INFO", "DATA_DIR": "./data" } } } } ``` #### 7.1.2 インストール手順 ```bash # 1. プロジェクト作成 uv init mcp_f0_server cd mcp_f0_server # 2. 依存関係インストール uv add "mcp[cli]" # 3. サーバー実装配置 cp mcp_f0_server.py ./ # 4. 権限設定（macOS/Linux） chmod +x mcp_f0_server.py # 5. Claude Desktopへの登録 uv run mcp install mcp_f0_server.py --name "F0 Random Generator" # 6. 動作確認 uv run mcp dev mcp_f0_server.py ``` ### 7.2 環境変数設定 | 変数名 | デフォルト値 | 説明 | |--------|------------|------| | MCP_LOG_LEVEL | INFO | ログレベル | | MCP_DATA_DIR | ./data | データ保存ディレクトリ | | MCP_MAX_NUMBERS | 10000 | 最大生成数制限 | | MCP_ENABLE_STATS | true | 統計機能の有効化 | --- ## 8. 監視と運用 ### 8.1 ログ管理 ```python # ログ設定 LOGGING_CONFIG = { "version": 1, "handlers": { "file": { "class": "logging.handlers.RotatingFileHandler", "filename": "mcp_f0_server.log", "maxBytes": 10485760, # 10MB "backupCount": 5, "formatter": "detailed" }, "console": { "class": "logging.StreamHandler", "formatter": "simple" } }, "formatters": { "detailed": { "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s" }, "simple": { "format": "%(levelname)s - %(message)s" } } } ``` ### 8.2 メトリクス収集 | メトリクス | 収集方法 | アラート閾値 | |-----------|---------|------------| | リクエスト数 | カウンター | - | | レスポンスタイム | ヒストグラム | >500ms | | エラー率 | レート計算 | >5% | | メモリ使用量 | システム監視 | >500MB | ### 8.3 トラブルシューティング #### 8.3.1 一般的な問題と解決策 | 問題 | 症状 | 解決策 | |-----|-----|--------| | 接続失敗 | "Server not found" | パス設定確認、権限チェック | | タイムアウト | レスポンスなし | ログ確認、プロセス再起動 | | データ不整合 | 誤った結果 | キャッシュクリア、再初期化 | #### 8.3.2 デバッグ手順 ```bash # 1. MCPインスペクターで確認 uv run mcp dev mcp_f0_server.py # 2. ログファイル確認（macOS/Linux） tail -f ~/Library/Logs/Claude/mcp-server-f0_random.log # 3. 手動実行テスト uv run mcp_f0_server.py # 4. 設定検証 uv run mcp validate-config ``` --- ## 9. セキュリティ考慮事項 ### 9.1 入力サニタイゼーション ```python def validate_input(value: Any, schema: Type[BaseModel]) -> BaseModel: """入力値の検証とサニタイゼーション""" try: return schema.model_validate(value) except ValidationError as e: logger.warning(f"Invalid input: {e}") raise MCPServerError("Invalid input parameters") ``` ### 9.2 ファイルアクセス制御 ```python ALLOWED_EXTENSIONS = {'.csv', '.json', '.txt'} BASE_DATA_DIR = Path("./data").resolve() def validate_filepath(filepath: str) -> Path: """ファイルパスの安全性検証""" path = Path(filepath).resolve() # ディレクトリトラバーサル防止 if not path.is_relative_to(BASE_DATA_DIR): raise SecurityError("Access denied") # 拡張子チェック if path.suffix not in ALLOWED_EXTENSIONS: raise SecurityError("Invalid file type") return path ``` ### 9.3 レート制限 ```python from asyncio import Semaphore class RateLimiter: def __init__(self, max_requests: int = 100): self.semaphore = Semaphore(max_requests) async def __aenter__(self): await self.semaphore.acquire() async def __aexit__(self, *args): self.semaphore.release() rate_limiter = RateLimiter() @mcp.tool() async def limited_operation(ctx: Context): async with rate_limiter: # 処理実装 pass ``` --- ## 10. 将来の拡張計画 ### 10.1 短期計画（3ヶ月） - [ ] Streamable HTTP transport対応 - [ ] バッチ処理機能の追加 - [ ] 統計分析機能の拡張 - [ ] プロンプトテンプレートの充実 ### 10.2 中期計画（6ヶ月） - [ ] FGDBシステムとの完全統合 - [ ] 他の関数ブロックのMCP化 - [ ] OAuth認証実装 - [ ] WebUIダッシュボード ### 10.3 長期計画（1年） - [ ] マルチテナント対応 - [ ] 分散処理対応 - [ ] MLパイプライン統合 - [ ] エンタープライズ機能 --- ## 11. リスク管理 ### 11.1 技術的リスク | リスク | 発生確率 | 影響度 | 対策 | |--------|---------|--------|------| | MCP仕様変更 | 中 | 高 | 定期的な仕様確認、バージョン管理 | | パフォーマンス劣化 | 低 | 中 | プロファイリング、最適化 | | 依存関係の脆弱性 | 中 | 高 | 定期的な更新、セキュリティ監査 | ### 11.2 運用リスク | リスク | 発生確率 | 影響度 | 対策 | |--------|---------|--------|------| | データ損失 | 低 | 高 | 定期バックアップ、冗長化 | | 設定ミス | 中 | 中 | 設定検証ツール、ドキュメント整備 | | リソース枯渇 | 低 | 中 | リソース監視、自動スケーリング | --- ## 12. 承認と合意事項 ### 12.1 ステークホルダー承認 | 役割 | 氏名 | 承認日 | 署名 | |------|------|--------|------| | プロジェクトマネージャー | - | - | - | | 技術リード | - | - | - | | QAリード | - | - | - | | セキュリティ責任者 | - | - | - | ### 12.2 前提条件 1. Python 3.8以上の実行環境が利用可能 2. MCP SDK最新版へのアクセス権限 3. Claude Desktopがインストール済み 4. 開発・テスト環境の提供 ### 12.3 制約事項 1. ローカル実行環境に限定（初期バージョン） 2. 同期的なファイルI/O処理 3. 単一プロセスでの実行 4. 英語ドキュメントのみ（初期） --- ## 付録A: 用語集 | 用語 | 定義 | |-----|------| | MCP | Model Context Protocol - LLMアプリケーション向け標準プロトコル | | FastMCP | MCPサーバー構築用Pythonフレームワーク | | stdio | Standard Input/Output - プロセス間通信方式 | | Tool | MCPでLLMに公開される実行可能な関数 | | Resource | MCPで提供されるデータアクセスポイント | | Prompt | MCPで定義されるインタラクションテンプレート | | FGDB | Functional Graph Database - 関数実行履歴管理システム | | Claude Desktop | Anthropic社のデスクトップAIアシスタント | --- ## 付録B: 参考資料 ### B.1 公式ドキュメント - [Model Context Protocol Specification](https://spec.modelcontextprotocol.io) - [MCP Python SDK Documentation](https://github.com/modelcontextprotocol/python-sdk) - [FastMCP Framework Guide](https://modelcontextprotocol.io/docs/tools/fastmcp) - [Claude Desktop Integration](https://modelcontextprotocol.io/docs/tools/claude-desktop) ### B.2 関連仕様 - [JSON-RPC 2.0 Specification](https://www.jsonrpc.org/specification) - [OAuth 2.1 Draft](https://datatracker.ietf.org/doc/draft-ietf-oauth-v2-1/) - [RFC 9728 - Protected Resource Metadata](https://datatracker.ietf.org/doc/rfc9728/) ### B.3 内部ドキュメント - FGDB システム仕様書 - f0_make_randomvalues 要件定義書（specification.md） --- ## 付録C: サンプルコード ### C.1 最小実装例 ```python #!/usr/bin/env python3 """ Minimal MCP server implementation for f0_make_randomvalues """ from mcp.server.fastmcp import FastMCP import random from typing import List # Create MCP server instance mcp = FastMCP(name="f0_random_minimal", version="1.0.0") @mcp.tool() def generate_random(count: int = 5) -> List[int]: """Generate random numbers""" return [random.randint(0, 999) for _ in range(count)] if __name__ == "__main__": mcp.run() ``` ### C.2 完全実装例（抜粋） ```python #!/usr/bin/env python3 """ Complete MCP server implementation with all features """ from mcp.server.fastmcp import FastMCP, Context from pydantic import BaseModel, Field from typing import List, Optional from pathlib import Path import random import csv import json from datetime import datetime from contextlib import asynccontextmanager from dataclasses import dataclass import logging # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Data models class RandomNumberResult(BaseModel): """Structured output for random number generation""" numbers: List[int] = Field(description="Generated random numbers") count: int = Field(description="Number of values generated") min_value: int = Field(description="Minimum value") max_value: int = Field(description="Maximum value") timestamp: str = Field(description="Generation timestamp") seed: Optional[int] = Field(default=None, description="Random seed used") # Application context @dataclass class AppContext: data_dir: Path current_data: Optional[List[int]] = None stats_cache: dict = None # Lifespan management @asynccontextmanager async def app_lifespan(server: FastMCP): """Manage server lifecycle""" # Initialize on startup data_dir = Path("./data") data_dir.mkdir(exist_ok=True) context = AppContext( data_dir=data_dir, stats_cache={} ) logger.info(f"MCP Server '{server.name}' initialized") logger.info(f"Data directory: {data_dir.absolute()}") try: yield context finally: # Cleanup on shutdown logger.info(f"MCP Server '{server.name}' shutting down") # Create server with lifespan mcp = FastMCP( name="f0_random_server", version="1.0.0", lifespan=app_lifespan ) # Tool implementations @mcp.tool() async def generate_random_numbers( ctx: Context, count: int = 5, min_value: int = 0, max_value: int = 999, seed: Optional[int] = None ) -> RandomNumberResult: """ Generate random numbers with specified parameters. Args: count: Number of random values to generate (default: 5) min_value: Minimum value (default: 0) max_value: Maximum value (default: 999) seed: Random seed for reproducibility (optional) Returns: Structured result containing generated numbers and metadata """ # Log operation await ctx.info(f"Generating {count} random numbers [{min_value}, {max_value}]") # Validate inputs if count <= 0 or count > 10000: raise ValueError("Count must be between 1 and 10000") if min_value >= max_value: raise ValueError("min_value must be less than max_value") # Set seed if provided if seed is not None: random.seed(seed) await ctx.debug(f"Using seed: {seed}") # Generate numbers numbers = [random.randint(min_value, max_value) for _ in range(count)] # Store in context ctx.request_context.lifespan_context.current_data = numbers # Create result result = RandomNumberResult( numbers=numbers, count=count, min_value=min_value, max_value=max_value, timestamp=datetime.now().isoformat(), seed=seed ) await ctx.info(f"Successfully generated {count} random numbers") return result # Resource implementations @mcp.resource("random://current") async def get_current_data(ctx: Context) -> str: """Get the most recently generated random numbers""" app_ctx = ctx.request_context.lifespan_context if app_ctx.current_data is None: return json.dumps({"error": "No data generated yet"}) return json.dumps({ "numbers": app_ctx.current_data, "count": len(app_ctx.current_data) }, indent=2) # Prompt implementations @mcp.prompt(title="Generate Test Dataset") def generate_test_dataset( purpose: str = "testing", size: str = "small" ) -> str: """Generate a prompt for creating test datasets""" sizes = { "small": 10, "medium": 100, "large": 1000 } count = sizes.get(size, 10) return f""" Please generate a random dataset for {purpose}. Requirements: - Size: {size} ({count} numbers) - Range: 0-999 - Format: CSV Use the generate_random_numbers tool with count={count}, then save the results. """ # Entry point if __name__ == "__main__": import sys # Check for debug mode if "--debug" in sys.argv: logging.getLogger().setLevel(logging.DEBUG) logger.debug("Debug mode enabled") # Run server logger.info("Starting MCP server...") mcp.run() ``` --- ## 改訂履歴 | バージョン | 日付 | 変更内容 | 作成者 | |-----------|------|---------|--------| | 1.0 | 2025-01-26 | 初版作成 | MCP開発チーム | --- *本仕様書は、f0_make_randomvalues関数ブロックのMCPサーバー実装に関する要件定義および詳細仕様を記載したものである。*