JVLink MCP Server

Instructions

Implementation Reference

• src/jvlink_mcp_server/server.py:663-666 (handler)

  The tool registration and the wrapper function `get_table_sample_data` that calls the implementation in `sample_data_provider`.

  def get_table_sample_data(table_name: str, num_rows: int = 5) -> dict:
      """テーブルのサンプルデータを取得（データ形式理解用）"""
      with DatabaseConnection() as db:
          return _get_sample_data(db, table_name=table_name, num_rows=num_rows)

• src/jvlink_mcp_server/database/sample_data_provider.py:31-115 (handler)

  The core logic implementation of `get_sample_data` which retrieves and formats table sample data.

  def get_sample_data(
      db_connection,
      table_name: str,
      num_rows: int = 5,
      where_clause: Optional[str] = None,
      use_cache: bool = True
  ) -> Dict[str, Any]:
      """テーブルからサンプルデータを取得
  
      Args:
          db_connection: DatabaseConnectionインスタンス
          table_name: テーブル名
          num_rows: 取得行数（デフォルト5行）
          where_clause: 追加のWHERE条件（例: "JyoCD = '05'"）
          use_cache: キャッシュを使用するか
  
      Returns:
          dict: {
              'table_name': テーブル名,
              'columns': カラム名リスト,
              'sample_rows': サンプルデータ（リスト形式）,
              'column_info': 重要カラムの説明,
              'data_format_notes': データ形式の注意事項
          }
      """
      # num_rows上限チェック
      num_rows = min(max(1, num_rows), 100)
  
      # テーブル名のホワイトリスト検証
      valid_tables = db_connection.get_tables()
      if table_name not in valid_tables:
          return {
              "table_name": table_name,
              "error": f"テーブル '{table_name}' は存在しません。有効なテーブル: {valid_tables}",
              "columns": [],
              "sample_rows": [],
          }
  
      cache_key = f"{table_name}_{num_rows}_{where_clause}"
  
      if use_cache and cache_key in _sample_data_cache:
          return _sample_data_cache[cache_key]
  
      # 重要カラムを優先して取得（ホワイトリスト検証）
      important_cols = IMPORTANT_COLUMNS.get(table_name, [])
  
      if important_cols:
          # カラム名をスキーマ情報でホワイトリスト検証
          try:
              schema_df = db_connection.get_table_schema(table_name)
              valid_columns = set(schema_df["column_name"].tolist())
              verified_cols = [c for c in important_cols if c in valid_columns]
              columns_str = ", ".join(verified_cols) if verified_cols else "*"
          except Exception:
              columns_str = "*"
      else:
          columns_str = "*"
  
      # SQL構築（where_clauseは無視 - SQLインジェクション対策）
      sql = f"SELECT {columns_str} FROM {table_name}"
  
      # 結果データがあるレコードを優先（NL_SEの場合）
      if table_name == "NL_SE":
          sql += " WHERE KakuteiJyuni IS NOT NULL AND KakuteiJyuni != ''"
  
      sql += f" LIMIT {num_rows}"
  
      try:
          df = db_connection.execute_safe_query(sql)
  
          result = {
              "table_name": table_name,
              "columns": df.columns.tolist(),
              "sample_rows": df.to_dict(orient="records"),
              "num_rows": len(df),
              "column_info": _get_column_info(table_name),
              "data_format_notes": _get_data_format_notes(table_name),
          }
  
          if use_cache:
              _sample_data_cache[cache_key] = result
  
          return result
  
      except Exception as e: