load_cpg
Loads a Code Property Graph (CPG) from a specified file path for code review and security analysis. Ensures the correct CPG is active in the Joern MCP Server.
Instructions
Loads a CPG from a file if the cpg is not loaded or the cpg is not the same as the filepath.
Args:
cpg_filepath (str): The path to the CPG file, the filepath is absolute path.
Returns:
str: True if the CPG is loaded successfully, False otherwise.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| cpg_filepath | Yes |
Implementation Reference
- server_tools.py:21-35 (handler)The handler function for the load_cpg tool, decorated with @joern_mcp.tool() for registration in the MCP server. It sends a remote query to the Joern server to load the specified CPG file and returns the extracted result.@joern_mcp.tool() def load_cpg(cpg_filepath: str) -> str: """ Loads a CPG from a file if the cpg is not loaded or the cpg is not the same as the filepath. Args: cpg_filepath (str): The path to the CPG file, the filepath is absolute path. Returns: str: True if the CPG is loaded successfully, False otherwise. """ # return extract_value(joern_remote(f'val cpg = CpgLoader.load("{cpg_filepath}")')) return extract_value(joern_remote(f'load_cpg("{cpg_filepath}")'))
- server.py:38-71 (helper)Helper function joern_remote used by load_cpg to execute the actual Joern query over HTTP to the server endpoint.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
- server.py:95-106 (registration)The generate() function that dynamically executes server_tools.py, thereby registering all tools including load_cpg via their decorators.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()