JVM MCP Server

Instructions

Implementation Reference

• src/jvm_mcp_server/server.py:120-182 (handler)

  Primary handler and registration for the get_thread_info tool. Validates PID input, executes JstackCommand to dump threads, and returns structured response with threads list, count, and metadata.
  @self.mcp.tool() def get_thread_info(pid: str = "") -> Dict: """获取指定进程的线程信息 Args: pid (str): 进程ID，使用字符串形式（如："12345"）。 支持十进制和十六进制格式。 空字符串将返回错误信息。 Returns: Dict: 包含线程信息的字典，包含以下字段： - threads (List[Dict]): 线程信息列表 - thread_count (int): 线程数量 - 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 { "threads": [], "thread_count": 0, "raw_output": "", "timestamp": time.time(), "success": False, "error": "Invalid process ID" } except ValueError as e: return { "threads": [], "thread_count": 0, "raw_output": "", "timestamp": time.time(), "success": False, "error": str(e) } cmd = JstackCommand(self.executor, JstackFormatter()) result = cmd.execute(str(validated_pid)) if not result.get('success', False): return { "threads": [], "thread_count": 0, "raw_output": result.get('output', ''), "timestamp": time.time(), "success": False, "error": result.get('error', 'Failed to execute jstack command') } threads = result.get('threads', []) # 返回格式化后的结果，包含 threads 字段 return { "threads": threads, "thread_count": len(threads), "raw_output": result.get('output', ''), "timestamp": time.time(), "success": True, "error": None }
• src/jvm_mcp_server/native/tools/jstack.py:25-141 (helper)

  Core helper that parses raw jstack output into structured thread objects including name, nid, state, stack_trace, and locks.
  class JstackFormatter(OutputFormatter): """JStack输出格式化器""" def format(self, result: CommandResult) -> Dict[str, Any]: """格式化jstack命令输出 Args: result: 命令执行结果 Returns: Dict[str, Any]: 格式化后的结果，包含线程信息 """ if not result.success: return { "success": False, "error": result.error, "timestamp": result.timestamp.isoformat() } threads: List[Dict[str, Any]] = [] current_thread: Optional[Dict[str, Any]] = None in_synchronizers = False for line in result.output.splitlines(): line = line.strip() if not line: continue # 新线程的开始 if line.startswith('"'): if current_thread: threads.append(current_thread) # 解析线程名和状态 # 格式: "thread-name" #id prio=5 os_prio=31 cpu=64.58ms elapsed=1.32s tid=0x00007f9a8d00e000 nid=0x2c03 waiting on condition name_end = line.rfind('"') if name_end > 0: thread_name = line[1:name_end] rest_line = line[name_end+1:].strip() # 解析线程ID和nid thread_id = None nid = None priority = None # 提取 #id if rest_line.startswith('#') or ' #' in rest_line: if rest_line.startswith('#'): id_part = rest_line[1:].split()[0] # 去掉开头的# else: id_part = rest_line.split(' #')[1].split()[0] try: thread_id = int(id_part) except ValueError: pass # 提取 nid (native thread id) if ' nid=' in rest_line: nid_part = rest_line.split(' nid=')[1].split()[0] nid = nid_part # 保持原始格式（通常是十六进制） # 提取优先级 if ' prio=' in rest_line: prio_part = rest_line.split(' prio=')[1].split()[0] try: priority = int(prio_part) except ValueError: pass current_thread = { "name": thread_name, "thread_id": thread_id, "nid": nid, "priority": priority, "state": "unknown", # 状态将在下一行更新 "stack_trace": [], "locks": [] } in_synchronizers = False # 解析线程状态 elif line.startswith('java.lang.Thread.State:'): if current_thread: current_thread["state"] = line.split(':', 1)[1].strip() in_synchronizers = False # 同步器信息开始 elif line.startswith('Locked synchronizers:'): in_synchronizers = True continue # 锁信息 elif line.startswith('- '): if current_thread: if in_synchronizers: if not line.startswith('- None'): # 跳过 "- None" current_thread["locks"].append(line.strip()) elif 'locked' in line or 'waiting to lock' in line or 'parking to wait' in line: current_thread["locks"].append(line.strip()) # 堆栈信息 elif line.startswith('at '): if current_thread: current_thread["stack_trace"].append(line.strip()) in_synchronizers = False # 添加最后一个线程 if current_thread: threads.append(current_thread) return { "success": True, "threads": threads, "thread_count": len(threads), "execution_time": result.execution_time, "timestamp": result.timestamp.isoformat() }
• src/jvm_mcp_server/server.py:63-98 (helper)

  Helper function used by get_thread_info (and other tools) to validate and convert PID strings to integers, supporting decimal and hex formats.
  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)}")