JVM MCP Server

Instructions

Implementation Reference

• src/jvm_mcp_server/server.py:226-266 (handler)

  Handler and registration for get_memory_info tool. Validates the PID input, executes JmapCommand with JmapHeapFormatter to retrieve JVM heap memory information, and formats the response.
  @self.mcp.tool() def get_memory_info(pid: str = "") -> Dict: """获取内存使用情况 Args: pid (str): 进程ID，使用字符串形式（如："12345"）。 支持十进制和十六进制格式。 空字符串将返回错误信息。 Returns: Dict: 包含内存信息的字典，包含以下字段： - raw_output (str): 原始输出 - timestamp (float): 时间戳 - success (bool): 是否成功 - error (Optional[str]): 错误信息 """ try: validated_pid = self._validate_and_convert_id(pid if pid else None, "process ID") if validated_pid is None: return { "raw_output": "", "timestamp": time.time(), "success": False, "error": "Invalid process ID" } except ValueError as e: return { "raw_output": "", "timestamp": time.time(), "success": False, "error": str(e) } cmd = JmapCommand(self.executor, JmapHeapFormatter()) result = cmd.execute(str(validated_pid)) return { "raw_output": result.get('output', ''), "timestamp": time.time(), "success": result.get('success', False), "error": result.get('error') }
• src/jvm_mcp_server/native/tools/jmap.py:132-148 (helper)

  Formatter used by the get_memory_info handler to process the raw output from JmapCommand into a standardized dictionary format.
  class JmapHeapFormatter(OutputFormatter): """Jmap堆内存概要格式化器（仅文本输出）""" def format(self, result: CommandResult) -> Dict[str, Any]: if not result.success: return { "success": False, "error": result.error, "timestamp": result.timestamp.isoformat() } return { "success": True, "output": result.output, "execution_time": result.execution_time, "timestamp": result.timestamp.isoformat() }
• src/jvm_mcp_server/native/tools/jmap.py:16-131 (helper)

  JmapCommand class used in get_memory_info to execute the appropriate jmap (or jhsdb) command for heap information based on JDK version.
  class JmapCommand(BaseCommand): """Jmap命令实现""" def __init__(self, executor, formatter): super().__init__(executor, formatter) self.timeout = 60 # 设置默认超时时间为60秒 self._jdk_version = None # 缓存 JDK 版本 def _get_jdk_version(self) -> int: """获取 JDK 主版本号""" if self._jdk_version is not None: return self._jdk_version try: # 检查是否为远程执行器 from ..base import NativeCommandExecutor if isinstance(self.executor, NativeCommandExecutor) and self.executor.ssh_host: # 远程执行 java -version result = self.executor.run('java -version', timeout=10) version_output = result.error if result.error else result.output else: # 本地执行 result = subprocess.run(['java', '-version'], capture_output=True, text=True, timeout=10) version_output = result.stderr # java -version 输出到 stderr # 解析版本号，支持多种格式 # 格式1: "openjdk version "11.0.12" 2021-07-20" # 格式2: "java version "1.8.0_291"" # 格式3: "openjdk version "17.0.15" 2025-04-15 LTS" version_patterns = [ r'version "1\.(\d+)', # 匹配 "1.8.0_291"，优先放前面 r'version "(\d+)', # 匹配 "11.0.12" 或 "17.0.15" ] for pattern in version_patterns: version_match = re.search(pattern, version_output) if version_match: version_str = version_match.group(1) if pattern == r'version "1\.(\d+)': # 对于 "1.8" 格式，返回 8 self._jdk_version = int(version_str) else: # 对于 "11" 或 "17" 格式，直接返回 self._jdk_version = int(version_str) break else: # 如果无法解析，假设是低版本 self._jdk_version = 8 except Exception as e: # 如果无法获取版本，假设是低版本 self._jdk_version = 8 return self._jdk_version def _is_modern_jdk(self) -> bool: """判断是否为现代 JDK (9+)""" return self._get_jdk_version() >= 9 def _test_jhsdb_availability(self) -> bool: """测试 jhsdb 命令是否可用""" try: from ..base import NativeCommandExecutor if isinstance(self.executor, NativeCommandExecutor) and self.executor.ssh_host: # 远程测试 result = self.executor.run('jhsdb --help', timeout=5) return result.success else: # 本地测试 result = subprocess.run(['jhsdb', '--help'], capture_output=True, text=True, timeout=5) return result.returncode == 0 except Exception: return False def get_command(self, pid: str, operation: JmapOperation = JmapOperation.HEAP, dump_file: Optional[str] = None, live_only: bool = False, *args, **kwargs) -> str: """获取jmap命令 Args: pid: 进程ID operation: 操作类型 dump_file: 转储文件路径（仅在dump操作时需要） live_only: 是否只统计存活对象 Returns: str: jmap命令字符串 """ # 验证 pid 参数 if not pid or not pid.strip(): raise ValueError("Process ID is required") try: int(pid) # 验证 pid 是否为有效数字 except ValueError: raise ValueError(f"Invalid process ID: {pid}") if operation == JmapOperation.HEAP: # 对于现代 JDK，优先尝试 jhsdb，如果不可用则回退到传统 jmap if self._is_modern_jdk() and self._test_jhsdb_availability(): return f'jhsdb jmap --heap --pid {pid}' else: return f'jmap -heap {pid}' elif operation == JmapOperation.HISTO: live_flag = " -live" if live_only else "" return f'jmap -histo{live_flag} {pid}' elif operation == JmapOperation.DUMP: if not dump_file: raise ValueError("dump_file is required for dump operation") live_flag = ":live" if live_only else "" return f'jmap -dump:format=b{live_flag},file={dump_file} {pid}' else: raise ValueError(f"Unsupported operation: {operation}")
• src/jvm_mcp_server/server.py:63-98 (helper)

  Helper method used in get_memory_info (and other tools) to validate and convert PID from string (decimal/hex) to int.
  def _validate_and_convert_id(self, value: Union[int, str, None], param_name: str = "ID") -> Optional[int]: """ 验证并转换ID参数，支持int和str类型的数字参数 Args: value: 要转换的值，可以是int、str或None param_name: 参数名称，用于错误信息 Returns: 转换后的整数值，如果输入为None则返回None Raises: ValueError: 如果无法转换为有效的整数 """ if value is None: return None if isinstance(value, int): return value if isinstance(value, str): # 去除前后空白字符 value = value.strip() if not value: return None try: # 支持十六进制格式（如0x2c03）和十进制格式 if value.lower().startswith('0x'): return int(value, 16) else: return int(value) except ValueError: raise ValueError(f"Invalid {param_name}: '{value}' cannot be converted to integer") raise ValueError(f"Invalid {param_name} type: expected int, str or None, got {type(value)}")