[codex] Add Binary Ninja Python execution tool

README.md

@@ -160,11 +160,14 @@ Your task is to analyze an unknown file which is currently open in Binary Ninja.
 
 ## Supported Capabilities
 
+MCP tools include safety annotations so clients can distinguish read-only inspection, additive analysis changes, and operations that may replace or delete state. Arbitrary Python execution is additionally marked as non-idempotent and open-world.
+
 The following table lists the available MCP functions for use:
 
 | Function                                                             | Description                                                                                                  |
 | -------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------ |
 | `decompile_function`                                                 | Decompile a specific function by name and return HLIL-like code with addresses.                              |
+| `execute_python(code, timeout_seconds)`                              | Execute code in a persistent Binary Ninja Python scripting context with access to `bv` and other console magic variables. |
 | `get_il(name_or_address, view, ssa)`                                 | Get IL for a function in `hlil`, `mlil`, or `llil` (SSA supported for MLIL/LLIL).                            |
 | `define_types`                                                       | Add type definitions from a C string type definition.                                                        |
 | `delete_comment`                                                     | Delete the comment at a specific address.                                                                    |
@@ -221,6 +224,8 @@ The following table lists the available MCP functions for use:
 
 These are the list of HTTP endpoints that can be called:
 
+- `/executePython` (POST): Execute JSON `code` in the Binary Ninja Python scripting context. Optional `timeout_seconds` is 30 by default and limited to 300.
+
 - `/allStrings`: All strings in one response.
 - `/formatValue?address=<addr>&text=<value>&size=<n>`: Convert and set a comment at an address.
 - `/getXrefsTo?address=<addr>`: Xrefs to address (code+data).

bridge/binja_mcp_bridge.py

@@ -134,6 +134,25 @@ def safe_post(endpoint: str, data: dict | str) -> str:
         return f"Request failed: {e!s}"
 
 
+def post_json(endpoint: str, data: dict, timeout: float = 5):
+    """Perform a POST and return parsed JSON, including structured HTTP errors."""
+    try:
+        response = requests.post(f"{binja_server_url}/{endpoint}", json=data, timeout=timeout)
+        response.encoding = "utf-8"
+        try:
+            result = response.json()
+        except Exception:
+            result = None
+        if response.ok:
+            return result
+        if isinstance(result, dict):
+            result.setdefault("error", f"HTTP {response.status_code}")
+            return result
+        return {"error": f"Error {response.status_code}: {response.text.strip()}"}
+    except Exception as e:
+        return {"error": f"Request failed: {e!s}"}
+
+
 @mcp.tool()
 def list_methods(offset: int = 0, limit: int = 100) -> list:
     """
@@ -764,6 +783,34 @@ def get_stack_frame_vars(function_identifier: str) -> list:
     return []
 
 
+@mcp.tool()
+def execute_python(code: str, timeout_seconds: float = 30) -> str:
+    """
+    Execute code in a persistent Binary Ninja Python scripting context.
+    Console magic variables such as bv, current_view, current_function, and here are available.
+    """
+    result = post_json(
+        "executePython",
+        {"code": code, "timeout_seconds": timeout_seconds},
+        timeout=timeout_seconds + 5,
+    )
+    if not isinstance(result, dict):
+        return "Error: no response"
+
+    sections = []
+    if result.get("output"):
+        sections.append(str(result["output"]))
+    if result.get("warnings"):
+        sections.append(f"Warnings:\n{result['warnings']}")
+    if result.get("errors"):
+        sections.append(f"Errors:\n{result['errors']}")
+    if result.get("error"):
+        sections.append(f"Error: {result['error']}")
+    if result.get("truncated"):
+        sections.append("[Output truncated after 1,000,000 characters]")
+    return "\n".join(sections) if sections else "(no output)"
+
+
 @mcp.tool()
 def format_value(address: str, text: str, size: int = 0) -> list:
     """

bridge/package.json

@@ -15,6 +15,7 @@
     ],
     "scripts": {
         "build": "tsc",
+        "test:annotations": "node scripts/test-tool-annotations.mjs",
         "start": "node dist/index.js",
         "dev": "tsx src/index.ts",
         "typecheck": "tsc --noEmit"

bridge/scripts/test-tool-annotations.mjs

@@ -0,0 +1,57 @@
+import assert from "node:assert/strict";
+import { fileURLToPath } from "node:url";
+
+import { Client } from "@modelcontextprotocol/sdk/client/index.js";
+import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
+
+const serverPath = fileURLToPath(new URL("../dist/index.js", import.meta.url));
+const transport = new StdioClientTransport({
+  command: process.execPath,
+  args: [serverPath],
+});
+const client = new Client({ name: "tool-annotations-test", version: "1.0.0" });
+
+try {
+  await client.connect(transport);
+  const { tools } = await client.listTools();
+  const missingAnnotations = tools.filter((tool) => !tool.annotations).map((tool) => tool.name);
+  const readOnly = tools.filter((tool) => tool.annotations?.readOnlyHint);
+  const additiveMutating = tools.filter(
+    (tool) => tool.annotations?.readOnlyHint === false && !tool.annotations?.destructiveHint,
+  );
+  const destructive = tools.filter((tool) => tool.annotations?.destructiveHint);
+
+  assert.equal(tools.length, 55);
+  assert.deepEqual(missingAnnotations, []);
+  assert.equal(readOnly.length, 37);
+  assert.deepEqual(
+    additiveMutating.map((tool) => tool.name).sort(),
+    ["make_function_at", "select_binary"],
+  );
+  assert.deepEqual(
+    destructive.map((tool) => tool.name).sort(),
+    [
+      "declare_c_type",
+      "define_types",
+      "delete_comment",
+      "delete_function_comment",
+      "execute_python",
+      "format_value",
+      "patch_bytes",
+      "rename_data",
+      "rename_function",
+      "rename_multi_variables",
+      "rename_single_variable",
+      "retype_variable",
+      "set_comment",
+      "set_function_comment",
+      "set_function_prototype",
+      "set_local_variable_type",
+    ],
+  );
+  assert.equal(tools.find((tool) => tool.name === "execute_python")?.annotations?.openWorldHint, true);
+  assert.equal(tools.find((tool) => tool.name === "execute_python")?.annotations?.idempotentHint, false);
+  assert.equal(tools.find((tool) => tool.name === "patch_bytes")?.annotations?.idempotentHint, true);
+} finally {
+  await client.close();
+}

bridge/src/client.ts

@@ -97,6 +97,28 @@ export class BinjaHttpClient {
     }
   }
 
+  /**
+   * Perform a POST request and return parsed JSON.
+   */
+  async postJson<T = unknown>(
+    endpoint: string,
+    data: Record<string, unknown>,
+    timeout?: number,
+  ): Promise<T | { error: string }> {
+    try {
+      const response = await this.client.post(endpoint, data, { timeout });
+      if (response.status >= 200 && response.status < 300) {
+        return response.data as T;
+      }
+      if (response.data && typeof response.data === "object" && "error" in response.data) {
+        return response.data as T;
+      }
+      return { error: `Error ${response.status}: ${response.statusText}` };
+    } catch (error) {
+      return { error: `Request failed: ${this.getErrorMessage(error)}` };
+    }
+  }
+
   private handleError(error: unknown, method: string, endpoint: string): string {
     const msg = this.getErrorMessage(error);
     return `Error: ${method} ${endpoint} failed: ${msg}`;