JVM MCP Server

Instructions

        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]): 错误信息
        

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()