Joern MCP Server

Instructions

@param method_full_name: The fully qualified name of the source method(e.g., com.android.nfc.NfcService$6.onReceive:void(android.content.Context,android.content.Intent))
@return: List of full name, name, signature and id of methods called by the source method

Implementation Reference

• server_tools.py:47-55 (handler)

  Handler function for 'get_method_callers' tool. Takes method_full_name, queries Joern server via joern_remote for callers, and returns extracted list using extract_list. The decorator @joern_mcp.tool() registers it as an MCP tool. The docstring provides input/output schema details.

  @joern_mcp.tool()
  def get_method_callers(method_full_name: str) -> list[str]:
      """Retrieves a list of methods that call the specified method
      
      @param method_full_name: The fully qualified name of the source method(e.g., com.android.nfc.NfcService$6.onReceive:void(android.content.Context,android.content.Intent))
      @return: List of full name, name, signature and id of methods called by the source method
      """
      responses = joern_remote(f'get_method_callers("{method_full_name}")')
      return extract_list(responses)

• server_tools.py:47-47 (registration)

  Decorator that registers the get_method_callers function as an MCP tool with the joern_mcp FastMCP instance.

  @joern_mcp.tool()

• server.py:38-71 (helper)

  Helper function joern_remote used by get_method_callers to execute the Joern query 'get_method_callers("{method_full_name}")' against the Joern server.

  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