JVM MCP Server

Instructions

Implementation Reference

• src/jvm_mcp_server/server.py:268-384 (handler)

  Primary handler implementation for the 'get_stack_trace' tool. Validates PID and parameters, executes JstackCommand to dump threads, filters by blocking status, specific thread ID, or top N threads, and returns structured thread information.
  @self.mcp.tool() def get_stack_trace(pid: str = "", thread_id: str = "", top_n: str = "5", find_blocking: bool = False, interval: str = "", show_all: bool = False) -> Dict: """获取线程堆栈信息 Args: pid (str): 进程ID，使用字符串形式（如："12345"） thread_id (str): 线程ID，使用字符串形式。支持十六进制（如："0x2c03"） top_n (str): 显示前N个线程，使用字符串形式（如："5"），默认值为"5" find_blocking (bool): 是否只查找阻塞线程（BLOCKED状态或等待锁的线程） interval (str): 采样间隔，使用字符串形式（如："1000"表示1秒） show_all (bool): 是否显示所有信息 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 { "raw_output": "", "timestamp": time.time(), "success": False, "error": "Invalid process ID" } validated_thread_id = self._validate_and_convert_id(thread_id if thread_id else None, "thread ID") validated_top_n = self._validate_and_convert_id(top_n if top_n else None, "top_n") # 设置默认值 if validated_top_n is None: validated_top_n = 5 except ValueError as e: return { "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', []) # 处理 find_blocking：筛选阻塞线程 if find_blocking: blocking_threads = [] for thread in threads: state = thread.get('state', '').upper() locks = thread.get('locks', []) # 检查是否为阻塞状态 is_blocked = ( 'BLOCKED' in state or any('waiting to lock' in lock.lower() for lock in locks) or any('parking to wait' in lock.lower() for lock in locks) ) if is_blocked: blocking_threads.append(thread) threads = blocking_threads # 处理 thread_id：筛选指定线程 if validated_thread_id is not None: target_threads = [] # 支持十进制和十六进制 target_nid_hex = hex(validated_thread_id) if isinstance(validated_thread_id, int) else str(validated_thread_id) target_nid_dec = str(validated_thread_id) for thread in threads: thread_nid = thread.get('nid', '') thread_tid = thread.get('thread_id') # 匹配nid（十六进制）或thread_id（十进制） if (thread_nid == target_nid_hex or thread_nid == target_nid_dec or thread_tid == validated_thread_id): target_threads.append(thread) threads = target_threads # 处理 top_n：取前N个线程 if validated_top_n is not None and validated_top_n > 0: threads = threads[:validated_top_n] # 返回格式化后的结果，包含 threads 字段 return { "threads": threads, "thread_count": len(threads), "raw_output": result.get('output', ''), "timestamp": time.time(), "success": True, "error": None }
• src/jvm_mcp_server/server.py:63-98 (helper)

  Helper method used by get_stack_trace to validate and convert string PIDs, thread_ids, top_n to integers, supporting hexadecimal format.
  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)}")
• src/jvm_mcp_server/server.py:268-268 (registration)

  Registration of the get_stack_trace tool using the FastMCP decorator.
  @self.mcp.tool()