Joern MCP Server

Instructions

@param class_full_name: The fully qualified name of the class
@return: The derived classes info of the class, including the full name, name and id

Implementation Reference

• server_tools.py:140-148 (handler)

  Handler function decorated with @joern_mcp.tool() implementing the tool logic: queries Joern server via joern_remote and parses response with extract_list to return list of derived classes.

  @joern_mcp.tool()
  def get_derived_classes_by_class_full_name(class_full_name:str) -> list[str]:
      """Get the derived classes of a class
      
      @param class_full_name: The fully qualified name of the class
      @return: The derived classes info of the class, including the full name, name and id
      """
      response = joern_remote(f'get_derived_classes_by_class_full_name("{class_full_name}")')
      return extract_list(response)

• server.py:95-106 (registration)

  Dynamic execution of server_tools.py via exec(), which runs the @joern_mcp.tool() decorators, registering the get_derived_classes_by_class_full_name tool among others.

  GENERATED_PY = os.path.join(SCRIPT_DIR, "server_tools.py")
  def generate():
      """Generate and execute additional server tools from server_tools.py file.
      
      This function reads the content of server_tools.py and executes it to add
      more functionality to the server.
      """
      with open(GENERATED_PY, "r") as f:
          code = f.read()
          exec(compile(code, GENERATED_PY, "exec"))
  
  generate()

• server.py:38-71 (helper)

  Core helper function that sends the Cypher-like query to the Joern server and returns the cleaned stdout response.

  def joern_remote(query):
      """
      Execute remote query and return results
      
      Parameters:
      query -- The query string to execute
      
      Returns:
      Returns the server response stdout content on success
      Returns None on failure, error message will be output to stderr
      """
      data = {"query": query}
      headers = {'Content-Type': 'application/json'}
  
      try:
          response = requests.post(
              f'http://{server_endpoint}/query-sync',
              data=json.dumps(data),
              headers=headers,
              auth=basic_auth,
              timeout=timeout
          )
          response.raise_for_status()  
          
          result = response.json()
          return remove_ansi_escape_sequences(result.get('stdout', ''))
          
      except requests.exceptions.RequestException as e:
          sys.stderr.write(f"Request Error: {str(e)}\n")
      except json.JSONDecodeError:
          sys.stderr.write("Error: Invalid JSON response\n")
      
      return None

• common_tools.py:38-77 (helper)

  Helper function used by the handler to parse the Joern server's Scala List response into a Python list of derived class names.

  def extract_list(input_str):    
      """Extract a list of elements from a string representation of a Scala List.
      
      Parameters:
      input_str -- The input string containing a Scala List representation
      
      Returns:
      A Python list containing the extracted elements with cleaned data
      """
      # Check if input is empty or None
      if not input_str:
          return []
      
      # Use regex to match List content
      list_pattern = r'List\((.*?)\)$'
      list_match = re.search(list_pattern, input_str, re.DOTALL)
      if not list_match:
          return []
          
      content = list_match.group(1).strip()
      
      # Try to match content within triple quotes
      triple_quote_pattern = r'"""(.*?)"""'
      triple_quote_matches = re.findall(triple_quote_pattern, content, re.DOTALL)
      
      if triple_quote_matches:
          return triple_quote_matches
      
      # If no triple-quoted content found, try to match content within regular quotes
      single_quote_pattern = r'"((?:\\.|[^"\\])*?)"'
      single_quote_matches = re.findall(single_quote_pattern, content, re.DOTALL)
      
      elements = []
      for item in single_quote_matches:
          if item.strip():
              # Handle escape characters
              cleaned = item.replace('\\"', '"').replace('\\\\', '\\')
              elements.append(cleaned)
      
      return elements