commit 65ed0398964d32f64d262e47e80f47d539e3390b
Author: WiseDev <83840010+wisedevik@users.noreply.github.com>
Date:   Fri Jul 31 13:23:46 2026 +0300

    Add dynamic multi-instance IDA MCP server and plugin
    
    Self-registering IDA plugin (auto port allocation, heartbeat-based
    instance registry) plus an MCP server exposing ~40 tools for
    decompilation, xrefs, type/struct editing, byte-level search and
    patching, Objective-C class recovery, and cross-instance comparison.

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..7a60b85
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,2 @@
+__pycache__/
+*.pyc
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..78e0c9e
--- /dev/null
+++ b/README.md
@@ -0,0 +1,127 @@
+# ida-mcp-server
+
+An MCP server that exposes a running IDA Pro session to Claude (or any other MCP
+client) over HTTP. Ships with an IDA plugin that turns a database into an
+HTTP API, and a small server that turns that API into MCP tools.
+
+## Why this exists
+
+The usual way to hook IDA up to an AI assistant is a fixed pair of ports:
+"IDA A" on 7777, "IDA B" on 7778, hardcoded into a script. That falls apart the
+moment you want a third binary open, or you're running IDA on a different
+machine over SSH and have to juggle port forwards by hand.
+
+This version drops the hardcoding. Each IDA instance finds a free port on its
+own, starts its own HTTP server, and registers itself with the MCP server over
+a heartbeat. Open as many IDA windows as you want, on the same machine or a
+remote one -- they show up automatically, and the MCP tools let you address a
+specific one by id or just let it pick the only one that's open.
+
+## How it's put together
+
+```
+ida_mcp_plugin.py   runs inside IDA, one instance per open database
+                     -> exposes decompilation, xrefs, renaming, etc. over HTTP
+ida-mcp-server.py    a small Starlette/uvicorn server
+                     -> turns that HTTP API into MCP tools over JSON-RPC (/sse)
+                     -> keeps a live registry of which IDA instances are open
+```
+
+The plugin and the server talk to each other in both directions: the plugin
+pushes heartbeats to the server so it knows what's alive, and the server pulls
+data from the plugin when a tool is called.
+
+## Setup
+
+1. Copy `ida_mcp_plugin.py` into IDA's plugins directory (on macOS, something
+   like `IDA Professional 9.1.app/Contents/MacOS/plugins/`).
+2. Install the server's dependencies and start it:
+
+   ```bash
+   pip install httpx starlette uvicorn
+   python ida-mcp-server.py
+   ```
+
+3. Open IDA. The plugin starts automatically, picks a port, and registers
+   itself with the server at `http://127.0.0.1:8888` by default.
+4. Point your MCP client at `http://<server-host>:8888/sse`.
+
+That's it for a single machine. No ports to configure, no restart dance when
+you open a second binary.
+
+### Multiple IDA instances
+
+Just open more IDA windows. Each one registers under its own id (derived from
+the input file name and process id) and shows up in `ida_list_instances`. Most
+tools take an optional `instance` argument -- skip it if only one IDA is open,
+or pass an id from `ida_list_instances` to target a specific one.
+
+### Running IDA on a different machine
+
+Set these before launching IDA on the remote box:
+
+- `IDA_MCP_HOST=0.0.0.0` -- so the plugin accepts connections from outside localhost
+- `IDA_MCP_ADVERTISE_HOST=<ip reachable from the server>` -- what the plugin tells the server to call it back on
+- `IDA_MCP_REGISTRY_URL=http://<server-ip>:8888` -- where to send heartbeats
+
+The server still has to be able to reach that IP and port -- this handles the
+port bookkeeping, not your network topology. If there's a firewall in the way
+you'll still need a tunnel, but you won't be assigning ports by hand anymore.
+
+### Reloading the plugin without restarting IDA
+
+If you edit `ida_mcp_plugin.py`, call the `ida_reload_plugin` tool instead of
+closing and reopening IDA. It re-reads the file from disk and swaps in the new
+code for anything that handles a request. The one thing that doesn't
+hot-reload is the heartbeat/registration thread, which keeps running with
+whatever code it started with until IDA actually restarts.
+
+## Security
+
+By default the server and plugin trust anything that can reach them, which is
+fine if everything stays on localhost. If you're exposing the server publicly
+(for example through ngrok, to use it from a hosted MCP client), set
+`IDA_MCP_TOKEN` to the same value on both the plugin's environment and the
+server's environment. That gates:
+
+- MCP tool calls against the server
+- instance registration from the plugin to the server
+- the OAuth handshake used by clients that require one (the token has to be
+  typed in during authorization; it isn't handed out automatically)
+
+Without a token, anyone who has the URL has full read/write access to whatever
+IDA session is open -- decompiling, renaming, patching, all of it. Don't skip
+the token if the server is reachable from outside your own machine.
+
+## What the tools can do
+
+Around 40 tools, roughly grouped as:
+
+- **Discovery**: list/search functions and strings, imports, segments, entry points, `ida_list_instances`
+- **Reading code**: decompile (single or batched), disassembly with honest pagination (it tells you if the output was truncated), xrefs, call graphs, a triage tool that ranks functions by how referenced or how large they are
+- **Types**: read/set a function's prototype, set a local variable's type, define structs/enums/unions in Local Types
+- **Writing to the database**: rename functions and locals, set comments (function-level or at a specific address), a persistent per-address note store that survives across sessions
+- **Binary-level**: raw byte reads, hex pattern search, immediate-value search, a heuristic vtable finder, byte patching, defining data items
+- **Objective-C**: class and method listing for Mach-O binaries, read from `_OBJC_CLASS_$_` symbols and `+[Class sel]`/`-[Class sel]` naming
+- **Comparing binaries**: decompile the same or different targets across two IDA instances side by side -- useful for diffing a patched build against the original, or a symboled build against a stripped one
+
+A few of these are marked experimental in their descriptions -- mostly around
+byte-pattern search and struct/type APIs that have moved between IDA versions
+in the past (IDA 9.x renamed or relocated a handful of functions this project
+already ran into and worked around). If one of them fails, the error message
+says what actually broke rather than a generic failure.
+
+## Known limitations
+
+- The Objective-C method scan relies on standard naming conventions
+  (`+[Class sel]`); it won't find much in binaries where those names have been
+  stripped or where the code is mostly Swift.
+- The vtable finder is a heuristic (runs of pointers into real functions in
+  non-executable segments) and can both miss real vtables and flag things that
+  aren't. Narrow it to a specific segment on large binaries or it can be slow.
+- Byte patching and struct/data creation write directly to the IDA database.
+  There's no undo built in beyond what IDA itself offers.
+
+## License
+
+No license file yet -- add one if you're planning to publish this more widely.
diff --git a/ida-mcp-server.py b/ida-mcp-server.py
new file mode 100644
index 0000000..7e2ae66
--- /dev/null
+++ b/ida-mcp-server.py
@@ -0,0 +1,939 @@
+"""
+IDA MCP Server -- a dynamic instance registry instead of hardcoded "IDA A / IDA B".
+
+How it works:
+  Each ida_mcp_plugin.py instance finds a free port on its own and sends a
+  heartbeat here every 10s (POST /instances/register). Open a third, a tenth
+  IDA -- it just shows up in the registry, no manual ports/SSH. Instances that
+  stop sending heartbeats for longer than INSTANCE_TTL seconds are dropped.
+
+  Tools take an optional "instance" argument:
+    - omitted, exactly one IDA open   -> that one is used
+    - omitted, several open           -> the most recently active one is used,
+                                          and the response says so explicitly
+    - given                           -> that specific instance is used
+  See what's currently open with the ida_list_instances tool.
+
+Run: python ida-mcp-server.py
+Requires: pip install httpx starlette uvicorn
+
+Security: if IDA_MCP_TOKEN is set (same value here and in the plugin's
+environment):
+  - MCP tool calls (/sse) require Authorization: Bearer <token>
+  - instance registration (/instances/register|unregister) requires the same token
+Without a token the server is fully open to anyone who can reach the port --
+fine for localhost-only use, but do NOT expose this over ngrok/the internet
+without one: anyone who gets the URL gets full control over your open IDA
+instances (decompiling, renaming, reading code).
+"""
+
+import json
+import os
+import secrets
+import time
+
+import httpx
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response, HTMLResponse
+from starlette.routing import Route
+
+SSE_PORT = int(os.environ.get("MCP_PORT", 8888))
+NGROK_URL = os.environ.get("NGROK_URL", "")
+AUTH_TOKEN = os.environ.get("IDA_MCP_TOKEN")
+INSTANCE_TTL = int(os.environ.get("IDA_MCP_INSTANCE_TTL", 30))
+
+http = httpx.AsyncClient(timeout=120.0)  # decompiling large functions can be slow
+
+# ─────────────────────────────────────────────
+#  Instance registry
+# ─────────────────────────────────────────────
+
+INSTANCES = {}  # id -> {host, port, file, pid, last_seen}
+AUTH_CODES = {}  # one-time OAuth codes -> expiry timestamp, issued only after the token page is filled correctly
+AUTH_CODE_TTL = 120
+
+
+def _live_instances():
+    now = time.time()
+    return {k: v for k, v in INSTANCES.items() if now - v["last_seen"] <= INSTANCE_TTL}
+
+
+def _check_token(request: Request) -> bool:
+    if not AUTH_TOKEN:
+        return True
+    return request.headers.get("Authorization", "") == f"Bearer {AUTH_TOKEN}"
+
+
+class ApiError(Exception):
+    """Exception with a machine-readable .code, so callers can branch on it
+    instead of parsing the error text."""
+
+    def __init__(self, message: str, code: str = "error"):
+        super().__init__(message)
+        self.code = code
+
+
+class ResolveError(ApiError):
+    pass
+
+
+def resolve_instance(instance_id):
+    """Returns (host, port, id, auto_picked). auto_picked=True means instance
+    wasn't passed explicitly and was chosen automatically (the most recently
+    active one) -- the caller should surface this in the response so the
+    choice stays visible instead of silent."""
+    live = _live_instances()
+    if instance_id:
+        if instance_id not in live:
+            available = ", ".join(live.keys()) or "(none)"
+            raise ResolveError(
+                f"Instance '{instance_id}' not found or not responding. Available: {available}. "
+                f"Call ida_list_instances to see what's currently open.",
+                code="instance_not_found",
+            )
+        v = live[instance_id]
+        return v["host"], v["port"], instance_id, False
+    if not live:
+        raise ResolveError(
+            "No IDA instances registered. Is the plugin running, and IDA_MCP_REGISTRY_URL "
+            f"pointing at this server (http://127.0.0.1:{SSE_PORT})?",
+            code="no_instances",
+        )
+    if len(live) == 1:
+        (only_id, v), = live.items()
+        return v["host"], v["port"], only_id, False
+    # several instances, none specified: use the most recently active one,
+    # but always report which one was picked instead of guessing silently
+    chosen_id, chosen = max(live.items(), key=lambda kv: kv[1]["last_seen"])
+    return chosen["host"], chosen["port"], chosen_id, True
+
+
+# ─────────────────────────────────────────────
+#  IDA helpers
+# ─────────────────────────────────────────────
+
+
+def _base(host, port):
+    return f"http://{host}:{port}"
+
+
+def _unwrap(r: httpx.Response):
+    if r.status_code >= 400:
+        code = "internal_error" if r.status_code >= 500 else "bad_request"
+        try:
+            body = r.json()
+            msg = body.get("error", r.text)
+            code = body.get("code", code)
+        except Exception:
+            msg = r.text
+        raise ApiError(f"IDA plugin error ({r.status_code}): {msg}", code=code)
+    return r.json()
+
+
+async def ida_get(host, port, path):
+    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} if AUTH_TOKEN else {}
+    r = await http.get(f"{_base(host, port)}{path}", headers=headers)
+    return _unwrap(r)
+
+
+async def ida_post(host, port, path, data):
+    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"} if AUTH_TOKEN else {}
+    r = await http.post(f"{_base(host, port)}{path}", json=data, headers=headers)
+    return _unwrap(r)
+
+
+# ─────────────────────────────────────────────
+#  Tools
+# ─────────────────────────────────────────────
+
+_INSTANCE_PARAM = {
+    "instance": {
+        "type": "string",
+        "description": "IDA instance id (see ida_list_instances). Omit if only one IDA is open.",
+    }
+}
+
+
+def _schema(props=None, required=None):
+    p = dict(_INSTANCE_PARAM)
+    p.update(props or {})
+    return {"type": "object", "properties": p, "required": required or []}
+
+
+TOOLS = [
+    {
+        "name": "ida_list_instances",
+        "description": (
+            "List currently open IDA instances (auto-registered by the plugin). Call this first "
+            "when unsure which binary/instance to target, or when a tool call errors about multiple "
+            "instances being open."
+        ),
+        "inputSchema": {"type": "object", "properties": {}},
+    },
+    {
+        "name": "ida_ping",
+        "description": "Check connectivity to an IDA instance; reports whether Hex-Rays decompiler is available.",
+        "inputSchema": _schema(),
+    },
+    {
+        "name": "ida_list_functions",
+        "description": "List functions with pagination. ALWAYS pass filter to avoid dumping huge lists into context.",
+        "inputSchema": _schema(
+            {
+                "filter": {"type": "string", "description": "Substring filter on function name — set this!"},
+                "limit": {"type": "integer", "default": 100},
+                "offset": {"type": "integer", "default": 0},
+            }
+        ),
+    },
+    {
+        "name": "ida_search_functions",
+        "description": "Search functions by substring in name. Prefer this over ida_list_functions for lookups.",
+        "inputSchema": _schema({"q": {"type": "string"}, "limit": {"type": "integer", "default": 50}}, ["q"]),
+    },
+    {
+        "name": "ida_search_strings",
+        "description": "Search decoded strings by substring.",
+        "inputSchema": _schema({"q": {"type": "string"}, "limit": {"type": "integer", "default": 30}}, ["q"]),
+    },
+    {
+        "name": "ida_get_name",
+        "description": "Resolve a symbol name (or 0xADDR) to its value — string contents or raw dword.",
+        "inputSchema": _schema({"name": {"type": "string"}}, ["name"]),
+    },
+    {
+        "name": "ida_get_imports",
+        "description": "List the import table, optionally filtered by substring.",
+        "inputSchema": _schema({"filter": {"type": "string"}}),
+    },
+    {
+        "name": "ida_get_segments",
+        "description": "List segments (name, address range, size, r/w/x permissions) — useful for orienting in an unfamiliar binary.",
+        "inputSchema": _schema(),
+    },
+    {
+        "name": "ida_get_entry_points",
+        "description": "List entry points of the binary.",
+        "inputSchema": _schema(),
+    },
+    {
+        "name": "ida_decompile",
+        "description": "Decompile a single function (Hex-Rays) by name or address (0x...).",
+        "inputSchema": _schema({"target": {"type": "string"}}, ["target"]),
+    },
+    {
+        "name": "ida_decompile_many",
+        "description": (
+            "Decompile several functions in one call (single round-trip instead of N) — use this instead "
+            "of calling ida_decompile in a loop when you already know the list of targets, e.g. after xrefs."
+        ),
+        "inputSchema": _schema(
+            {"targets": {"type": "array", "items": {"type": "string"}, "description": "Names or 0x addresses"}},
+            ["targets"],
+        ),
+    },
+    {
+        "name": "ida_get_disasm",
+        "description": (
+            "Raw disassembly listing for a function — use when Hex-Rays fails or asm-level detail is needed. "
+            "Response includes total_instructions/returned/truncated — if truncated is true, call again with a "
+            "higher count or a nonzero offset to page through the rest instead of assuming you've seen it all."
+        ),
+        "inputSchema": _schema(
+            {
+                "target": {"type": "string"},
+                "count": {"type": "integer", "default": 30},
+                "offset": {"type": "integer", "default": 0, "description": "Skip this many instructions before returning — for paging past a truncated result"},
+            },
+            ["target"],
+        ),
+    },
+    {
+        "name": "ida_xrefs_to",
+        "description": "Find all locations that reference an address/function (callers, data refs).",
+        "inputSchema": _schema({"target": {"type": "string"}}, ["target"]),
+    },
+    {
+        "name": "ida_xrefs_from",
+        "description": "Find all addresses a function references (callees).",
+        "inputSchema": _schema({"target": {"type": "string"}}, ["target"]),
+    },
+    {
+        "name": "ida_get_prototype",
+        "description": "Get the current type/prototype string of a function.",
+        "inputSchema": _schema({"target": {"type": "string"}}, ["target"]),
+    },
+    {
+        "name": "ida_set_prototype",
+        "description": 'Set a function\'s prototype, e.g. "int __fastcall foo(int a, char *b);".',
+        "inputSchema": _schema({"target": {"type": "string"}, "prototype": {"type": "string"}}, ["target", "prototype"]),
+    },
+    {
+        "name": "ida_set_lvar_type",
+        "description": (
+            'Set the type of a local variable inside a decompiled function, e.g. type="int" or "MyStruct *". '
+            "Experimental — Hex-Rays API varies slightly across IDA versions; check the error message if it fails."
+        ),
+        "inputSchema": _schema(
+            {"function": {"type": "string"}, "lvar_name": {"type": "string"}, "type": {"type": "string"}},
+            ["function", "lvar_name", "type"],
+        ),
+    },
+    {
+        "name": "ida_rename_function",
+        "description": "Rename a function.",
+        "inputSchema": _schema({"target": {"type": "string"}, "new_name": {"type": "string"}}, ["target", "new_name"]),
+    },
+    {
+        "name": "ida_rename_local",
+        "description": "Rename a local variable inside a decompiled function.",
+        "inputSchema": _schema(
+            {"function": {"type": "string"}, "old_name": {"type": "string"}, "new_name": {"type": "string"}},
+            ["function", "old_name", "new_name"],
+        ),
+    },
+    {
+        "name": "ida_set_comment",
+        "description": "Set a function-level comment.",
+        "inputSchema": _schema({"target": {"type": "string"}, "comment": {"type": "string"}}, ["target", "comment"]),
+    },
+    {
+        "name": "ida_get_notes",
+        "description": (
+            "Read the persistent AI scratchpad note stored at an address (survives across sessions/agents, "
+            "stored inside the idb). Use this to check if a function was already analyzed before re-doing work."
+        ),
+        "inputSchema": _schema({"target": {"type": "string"}}, ["target"]),
+    },
+    {
+        "name": "ida_set_notes",
+        "description": (
+            "Write a persistent AI scratchpad note at an address — e.g. your analysis conclusion for a "
+            "function, so future calls (or other agents) don't have to re-derive it. Empty string deletes it."
+        ),
+        "inputSchema": _schema({"target": {"type": "string"}, "note": {"type": "string"}}, ["target", "note"]),
+    },
+    {
+        "name": "ida_rebuild_name_cache",
+        "description": "Force-rebuild the plugin's internal name cache (rarely needed; use if get_name misses a symbol that clearly exists).",
+        "inputSchema": _schema(),
+    },
+    {
+        "name": "ida_get_bytes",
+        "description": "Read raw bytes at an address as hex — for shellcode, keys, magic constants, or anything not worth disassembling.",
+        "inputSchema": _schema({"target": {"type": "string"}, "size": {"type": "integer", "default": 64}}, ["target"]),
+    },
+    {
+        "name": "ida_find_pattern",
+        "description": (
+            'Search the whole binary for an IDA-style hex byte pattern, e.g. "48 8B ?? ??" (?? = wildcard byte). '
+            "Experimental — underlying IDA search API varies across versions."
+        ),
+        "inputSchema": _schema({"pattern": {"type": "string"}, "limit": {"type": "integer", "default": 50}}, ["pattern"]),
+    },
+    {
+        "name": "ida_patch_bytes",
+        "description": "Overwrite raw bytes at an address in the binary. Destructive — confirm with the user before calling.",
+        "inputSchema": _schema({"target": {"type": "string"}, "hex_bytes": {"type": "string", "description": "e.g. '90 90 90' for 3 NOPs"}}, ["target", "hex_bytes"]),
+    },
+    {
+        "name": "ida_make_data",
+        "description": "Define a data item (byte/word/dword/qword/struct) at an address. Experimental; struct requires struct_name from ida_list_local_types.",
+        "inputSchema": _schema(
+            {
+                "target": {"type": "string"},
+                "data_type": {"type": "string", "enum": ["byte", "word", "dword", "qword", "struct"]},
+                "struct_name": {"type": "string"},
+            },
+            ["target", "data_type"],
+        ),
+    },
+    {
+        "name": "ida_get_line_comment",
+        "description": "Read the comment at a specific address (disassembly-line level, distinct from the whole-function comment).",
+        "inputSchema": _schema({"target": {"type": "string"}, "repeatable": {"type": "boolean", "default": False}}, ["target"]),
+    },
+    {
+        "name": "ida_set_line_comment",
+        "description": "Set a comment at a specific address (disassembly-line level, distinct from ida_set_comment which is whole-function).",
+        "inputSchema": _schema(
+            {"target": {"type": "string"}, "comment": {"type": "string"}, "repeatable": {"type": "boolean", "default": False}},
+            ["target", "comment"],
+        ),
+    },
+    {
+        "name": "ida_top_functions",
+        "description": (
+            "Triage tool for an unfamiliar/stripped binary: list the most cross-referenced or largest functions "
+            "first, instead of guessing where to start."
+        ),
+        "inputSchema": _schema({"by": {"type": "string", "enum": ["xrefs", "size"], "default": "xrefs"}, "limit": {"type": "integer", "default": 20}}),
+    },
+    {
+        "name": "ida_call_graph",
+        "description": "Build a callers or callees graph up to N levels deep from a function, in one call instead of walking xrefs manually.",
+        "inputSchema": _schema(
+            {
+                "target": {"type": "string"},
+                "depth": {"type": "integer", "default": 2},
+                "direction": {"type": "string", "enum": ["callees", "callers"], "default": "callees"},
+            },
+            ["target"],
+        ),
+    },
+    {
+        "name": "ida_make_function",
+        "description": "Create a function at an address that auto-analysis missed.",
+        "inputSchema": _schema({"target": {"type": "string"}}, ["target"]),
+    },
+    {
+        "name": "ida_undefine",
+        "description": "Remove function/data definition at an address (undo a bad auto-analysis guess before redefining it).",
+        "inputSchema": _schema({"target": {"type": "string"}}, ["target"]),
+    },
+    {
+        "name": "ida_list_local_types",
+        "description": "List structs/enums currently defined in Local Types.",
+        "inputSchema": _schema({"limit": {"type": "integer", "default": 200}}),
+    },
+    {
+        "name": "ida_create_struct",
+        "description": 'Define a new struct in Local Types from a field list, e.g. fields=[{"name":"flag","type":"int"},{"name":"ptr","type":"void *"}].',
+        "inputSchema": _schema(
+            {
+                "name": {"type": "string"},
+                "fields": {"type": "array", "items": {"type": "object", "properties": {"name": {"type": "string"}, "type": {"type": "string"}}}},
+            },
+            ["name", "fields"],
+        ),
+    },
+    {
+        "name": "ida_get_objc_classes",
+        "description": (
+            "List Objective-C classes and methods recovered from the binary (by _OBJC_CLASS_$_ symbols and "
+            "+[Class sel]/-[Class sel] function naming). Use this first on any Mach-O/iOS/macOS target with ObjC "
+            "— it's often more useful than decompiling method-by-method."
+        ),
+        "inputSchema": _schema({"filter": {"type": "string"}, "limit": {"type": "integer", "default": 200}}),
+    },
+    {
+        "name": "ida_find_immediate",
+        "description": (
+            "Search all instructions for an immediate operand matching a given value (decimal or 0x-hex) — "
+            "e.g. find where a magic constant, size, or flag literal is used, unlike ida_find_pattern which "
+            "matches raw bytes regardless of instruction semantics."
+        ),
+        "inputSchema": _schema({"value": {"type": "string", "description": "e.g. '1337' or '0x539'"}, "limit": {"type": "integer", "default": 100}}, ["value"]),
+    },
+    {
+        "name": "ida_find_vtables",
+        "description": (
+            "Heuristic search for virtual-function-table-like arrays: runs of consecutive pointers in "
+            "non-executable segments where every pointer is the start of a real function. Useful for finding "
+            "C++ vtables. Experimental — heuristic, can have false positives/negatives; narrow with 'segment' "
+            "(e.g. '__const', '__data') on large binaries for speed."
+        ),
+        "inputSchema": _schema(
+            {
+                "segment": {"type": "string", "description": "Optional segment name substring to narrow the scan"},
+                "min_entries": {"type": "integer", "default": 2},
+                "limit": {"type": "integer", "default": 50},
+            }
+        ),
+    },
+    {
+        "name": "ida_create_local_type",
+        "description": (
+            'Define a raw C declaration (struct/enum/union/typedef) in Local Types, e.g. decl="enum Color { RED, GREEN, BLUE };". '
+            "More general than ida_create_struct — use this for enums/unions/typedefs, or structs with bitfields/nesting."
+        ),
+        "inputSchema": _schema({"decl": {"type": "string"}}, ["decl"]),
+    },
+    {
+        "name": "ida_xrefs_to_many",
+        "description": "Get xrefs_to for several targets in one call — e.g. right after ida_search_functions, instead of calling ida_xrefs_to in a loop.",
+        "inputSchema": _schema({"targets": {"type": "array", "items": {"type": "string"}}}, ["targets"]),
+    },
+    {
+        "name": "ida_reload_plugin",
+        "description": (
+            "Re-read the IDA plugin's Python code from disk without restarting IDA — use after the plugin file "
+            "itself was edited, so bug fixes/new endpoints take effect without a full IDA restart. Note: the "
+            "background heartbeat/registration thread keeps its old code until an actual IDA restart; only "
+            "request-handling logic hot-reloads."
+        ),
+        "inputSchema": _schema(),
+    },
+    {
+        "name": "ida_compare",
+        "description": (
+            "Decompile a target in two different IDA instances and return both side by side — e.g. to diff "
+            "a patched binary against the original, or a symboled build against a stripped one."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "instance_a": {"type": "string", "description": "First instance id (see ida_list_instances)"},
+                "target_a": {"type": "string"},
+                "instance_b": {"type": "string", "description": "Second instance id"},
+                "target_b": {"type": "string"},
+            },
+            "required": ["instance_a", "target_a", "instance_b", "target_b"],
+        },
+    },
+]
+
+
+# ─────────────────────────────────────────────
+#  Tool execution
+# ─────────────────────────────────────────────
+
+
+async def execute_tool(name: str, arguments: dict):
+    try:
+        if name == "ida_list_instances":
+            live = _live_instances()
+            return {
+                "count": len(live),
+                "instances": [
+                    {"id": k, "host": v["host"], "port": v["port"], "file": v["file"]}
+                    for k, v in live.items()
+                ],
+            }
+
+        if name == "ida_compare":
+            ha, pa, _, _ = resolve_instance(arguments["instance_a"])
+            hb, pb, _, _ = resolve_instance(arguments["instance_b"])
+            a = await ida_get(ha, pa, f"/decompile/{arguments['target_a']}")
+            b = await ida_get(hb, pb, f"/decompile/{arguments['target_b']}")
+            return {"instance_a": arguments["instance_a"], "instance_b": arguments["instance_b"], "a": a, "b": b}
+
+        host, port, inst_id, auto_picked = resolve_instance(arguments.get("instance"))
+        result = await _dispatch(name, arguments, host, port)
+        if isinstance(result, dict) and auto_picked:
+            result.setdefault(
+                "_instance_used", f"{inst_id} (auto-picked, multiple instances were open — pass 'instance' to target a specific one)"
+            )
+        return result
+
+    except ApiError as e:
+        return {"error": str(e), "code": e.code}
+    except httpx.ConnectError:
+        return {"error": "Cannot connect to the IDA instance. Is the plugin still running?", "code": "instance_unreachable"}
+    except KeyError as e:
+        return {"error": f"Missing required argument: {e}", "code": "bad_request"}
+    except Exception as e:
+        return {"error": str(e), "code": "internal_error"}
+
+
+async def _dispatch(name: str, arguments: dict, host, port):
+    if name == "ida_ping":
+        return await ida_get(host, port, "/ping")
+    elif name == "ida_list_functions":
+        f = arguments.get("filter", "")
+        limit = arguments.get("limit", 100)
+        offset = arguments.get("offset", 0)
+        return await ida_get(host, port, f"/functions?filter={f}&limit={limit}&offset={offset}")
+    elif name == "ida_search_functions":
+        return await ida_get(host, port, f"/functions/search?q={arguments['q']}&limit={arguments.get('limit', 50)}")
+    elif name == "ida_search_strings":
+        return await ida_get(host, port, f"/strings/search?q={arguments['q']}&limit={arguments.get('limit', 30)}")
+    elif name == "ida_get_name":
+        return await ida_get(host, port, f"/name/{arguments['name']}")
+    elif name == "ida_get_imports":
+        return await ida_get(host, port, f"/imports?filter={arguments.get('filter', '')}")
+    elif name == "ida_get_segments":
+        return await ida_get(host, port, "/segments")
+    elif name == "ida_get_entry_points":
+        return await ida_get(host, port, "/entry_points")
+    elif name == "ida_decompile":
+        return await ida_get(host, port, f"/decompile/{arguments['target']}")
+    elif name == "ida_decompile_many":
+        return await ida_post(host, port, "/decompile_batch", {"targets": arguments["targets"]})
+    elif name == "ida_get_disasm":
+        return await ida_get(
+            host, port, f"/disasm/{arguments['target']}?count={arguments.get('count', 30)}&offset={arguments.get('offset', 0)}"
+        )
+    elif name == "ida_xrefs_to":
+        return await ida_get(host, port, f"/xrefs_to/{arguments['target']}")
+    elif name == "ida_xrefs_from":
+        return await ida_get(host, port, f"/xrefs_from/{arguments['target']}")
+    elif name == "ida_get_prototype":
+        return await ida_get(host, port, f"/prototype/{arguments['target']}")
+    elif name == "ida_set_prototype":
+        return await ida_post(host, port, "/set_prototype", {"target": arguments["target"], "prototype": arguments["prototype"]})
+    elif name == "ida_set_lvar_type":
+        return await ida_post(
+            host,
+            port,
+            "/set_lvar_type",
+            {"function": arguments["function"], "lvar_name": arguments["lvar_name"], "type": arguments["type"]},
+        )
+    elif name == "ida_rename_function":
+        return await ida_post(host, port, "/rename_function", {"target": arguments["target"], "new_name": arguments["new_name"]})
+    elif name == "ida_rename_local":
+        return await ida_post(
+            host,
+            port,
+            "/rename_local",
+            {"function": arguments["function"], "old_name": arguments["old_name"], "new_name": arguments["new_name"]},
+        )
+    elif name == "ida_set_comment":
+        return await ida_post(host, port, "/set_comment", {"target": arguments["target"], "comment": arguments["comment"]})
+    elif name == "ida_get_notes":
+        return await ida_get(host, port, f"/notes/{arguments['target']}")
+    elif name == "ida_set_notes":
+        return await ida_post(host, port, "/notes", {"target": arguments["target"], "note": arguments["note"]})
+    elif name == "ida_rebuild_name_cache":
+        return await ida_get(host, port, "/name_cache/rebuild")
+    elif name == "ida_get_bytes":
+        return await ida_get(host, port, f"/bytes/{arguments['target']}?size={arguments.get('size', 64)}")
+    elif name == "ida_find_pattern":
+        return await ida_get(host, port, f"/find?pattern={arguments['pattern']}&limit={arguments.get('limit', 50)}")
+    elif name == "ida_patch_bytes":
+        return await ida_post(host, port, "/patch_bytes", {"target": arguments["target"], "hex_bytes": arguments["hex_bytes"]})
+    elif name == "ida_make_data":
+        return await ida_post(
+            host,
+            port,
+            "/make_data",
+            {"target": arguments["target"], "data_type": arguments["data_type"], "struct_name": arguments.get("struct_name")},
+        )
+    elif name == "ida_get_line_comment":
+        rep = "1" if arguments.get("repeatable") else "0"
+        return await ida_get(host, port, f"/line_comment/{arguments['target']}?repeatable={rep}")
+    elif name == "ida_set_line_comment":
+        return await ida_post(
+            host,
+            port,
+            "/line_comment",
+            {"target": arguments["target"], "comment": arguments["comment"], "repeatable": bool(arguments.get("repeatable", False))},
+        )
+    elif name == "ida_top_functions":
+        return await ida_get(host, port, f"/top_functions?by={arguments.get('by', 'xrefs')}&limit={arguments.get('limit', 20)}")
+    elif name == "ida_call_graph":
+        return await ida_get(
+            host,
+            port,
+            f"/call_graph/{arguments['target']}?depth={arguments.get('depth', 2)}&direction={arguments.get('direction', 'callees')}",
+        )
+    elif name == "ida_make_function":
+        return await ida_post(host, port, "/make_function", {"target": arguments["target"]})
+    elif name == "ida_undefine":
+        return await ida_post(host, port, "/undefine", {"target": arguments["target"]})
+    elif name == "ida_list_local_types":
+        return await ida_get(host, port, f"/local_types?limit={arguments.get('limit', 200)}")
+    elif name == "ida_create_struct":
+        return await ida_post(host, port, "/create_struct", {"name": arguments["name"], "fields": arguments["fields"]})
+    elif name == "ida_get_objc_classes":
+        f = arguments.get("filter", "")
+        return await ida_get(host, port, f"/objc_classes?filter={f}&limit={arguments.get('limit', 200)}")
+    elif name == "ida_find_immediate":
+        return await ida_get(host, port, f"/find_immediate?value={arguments['value']}&limit={arguments.get('limit', 100)}")
+    elif name == "ida_find_vtables":
+        seg = arguments.get("segment", "") or ""
+        return await ida_get(
+            host,
+            port,
+            f"/find_vtables?segment={seg}&min_entries={arguments.get('min_entries', 2)}&limit={arguments.get('limit', 50)}",
+        )
+    elif name == "ida_create_local_type":
+        return await ida_post(host, port, "/create_type", {"decl": arguments["decl"]})
+    elif name == "ida_xrefs_to_many":
+        return await ida_post(host, port, "/xrefs_to_batch", {"targets": arguments["targets"]})
+    elif name == "ida_reload_plugin":
+        return await ida_get(host, port, "/reload")
+
+    return {"error": f"Unknown tool: {name}", "code": "unknown_tool"}
+
+
+# ─────────────────────────────────────────────
+#  Instance registry endpoints (called by the plugin)
+# ─────────────────────────────────────────────
+
+
+async def handle_register_instance(request: Request):
+    if not _check_token(request):
+        return JSONResponse({"error": "unauthorized"}, status_code=401)
+    try:
+        body = await request.json()
+    except Exception:
+        return JSONResponse({"error": "invalid json"}, status_code=400)
+
+    inst_id = body.get("id")
+    host = body.get("host")
+    port = body.get("port")
+    if not inst_id or not host or not port:
+        return JSONResponse({"error": "id, host, port are required"}, status_code=400)
+
+    INSTANCES[inst_id] = {
+        "host": host,
+        "port": port,
+        "file": body.get("file", ""),
+        "pid": body.get("pid"),
+        "last_seen": time.time(),
+    }
+    return JSONResponse({"ok": True})
+
+
+async def handle_unregister_instance(request: Request):
+    if not _check_token(request):
+        return JSONResponse({"error": "unauthorized"}, status_code=401)
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+    INSTANCES.pop(body.get("id"), None)
+    return JSONResponse({"ok": True})
+
+
+async def handle_list_instances(request: Request):
+    if not _check_token(request):
+        return JSONResponse({"error": "unauthorized"}, status_code=401)
+    live = _live_instances()
+    return JSONResponse({"count": len(live), "instances": [{"id": k, **v} for k, v in live.items()]})
+
+
+# ─────────────────────────────────────────────
+#  MCP JSON-RPC
+# ─────────────────────────────────────────────
+
+
+def make_response(req_id, result):
+    return {"jsonrpc": "2.0", "id": req_id, "result": result}
+
+
+def make_error(req_id, code, message):
+    return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}}
+
+
+async def handle_jsonrpc(msg: dict):
+    method = msg.get("method")
+    params = msg.get("params", {})
+    req_id = msg.get("id")
+
+    if method == "initialize":
+        return make_response(
+            req_id,
+            {
+                "protocolVersion": "2024-11-05",
+                "capabilities": {"tools": {}},
+                "serverInfo": {"name": "ida-mcp", "version": "4.0.0"},
+            },
+        )
+    elif method in ("notifications/initialized", "initialized"):
+        return None
+    elif method == "tools/list":
+        return make_response(req_id, {"tools": TOOLS})
+    elif method == "tools/call":
+        result = await execute_tool(params.get("name"), params.get("arguments", {}))
+        is_error = isinstance(result, dict) and "error" in result
+        return make_response(
+            req_id,
+            {
+                "content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}],
+                "isError": is_error,
+            },
+        )
+    elif method == "ping":
+        return make_response(req_id, {})
+    else:
+        return make_error(req_id, -32601, f"Method not found: {method}")
+
+
+# ─────────────────────────────────────────────
+#  HTTP handlers
+# ─────────────────────────────────────────────
+
+
+async def handle_mcp_post(request: Request):
+    if request.method == "HEAD":
+        return Response(status_code=200)
+    if request.method == "GET":
+        return JSONResponse({"error": "use POST for JSON-RPC"}, status_code=400)
+    if not _check_token(request):
+        return JSONResponse(make_error(None, -32000, "unauthorized"), status_code=401)
+    try:
+        body = await request.json()
+    except Exception:
+        return JSONResponse(make_error(None, -32700, "Parse error"), status_code=400)
+
+    if isinstance(body, list):
+        responses = [r for r in [await handle_jsonrpc(m) for m in body] if r is not None]
+        return JSONResponse(responses)
+
+    result = await handle_jsonrpc(body)
+    if result is None:
+        return Response(status_code=202)
+    return JSONResponse(result)
+
+
+async def handle_health(request: Request):
+    return JSONResponse(
+        {
+            "status": "ok",
+            "server": "ida-mcp-dynamic",
+            "version": "4.0.0",
+            "instances": len(_live_instances()),
+            "auth_required": bool(AUTH_TOKEN),
+        }
+    )
+
+
+async def handle_oauth_protected_resource(request: Request):
+    return JSONResponse(
+        {
+            "resource": NGROK_URL,
+            "authorization_servers": [],
+            "bearer_methods_supported": ["header"],
+            "scopes_supported": [],
+        }
+    )
+
+
+async def handle_oauth_authorization_server(request: Request):
+    return JSONResponse(
+        {
+            "issuer": NGROK_URL,
+            "authorization_endpoint": f"{NGROK_URL}/authorize",
+            "token_endpoint": f"{NGROK_URL}/token",
+            "registration_endpoint": f"{NGROK_URL}/register",
+            # client_credentials would let a script skip the /authorize prompt entirely — not offered.
+            "grant_types_supported": ["authorization_code"],
+            "response_types_supported": ["code"],
+            "code_challenge_methods_supported": ["S256", "plain"],
+            "token_endpoint_auth_methods_supported": ["none"],
+            "scopes_supported": [],
+        }
+    )
+
+
+def _authorize_form_html(redirect_uri: str, state: str, error: str = "") -> str:
+    error_html = f'<p style="color:#c00">{error}</p>' if error else ""
+    return f"""<!doctype html><html><body style="font-family:sans-serif;max-width:420px;margin:80px auto">
+<h3>IDA MCP — enter access token</h3>
+{error_html}
+<form method="GET" action="/authorize">
+  <input type="hidden" name="redirect_uri" value="{redirect_uri}">
+  <input type="hidden" name="state" value="{state}">
+  <input type="password" name="token" placeholder="IDA_MCP_TOKEN" style="width:100%;padding:8px;margin:8px 0" autofocus>
+  <button type="submit" style="padding:8px 16px">Authorize</button>
+</form>
+</body></html>"""
+
+
+def _issue_code() -> str:
+    code = secrets.token_urlsafe(24)
+    AUTH_CODES[code] = time.time() + AUTH_CODE_TTL
+    return code
+
+
+def _redirect_with_code(redirect_uri: str, state: str) -> HTMLResponse:
+    code = _issue_code()
+    sep = "&" if "?" in redirect_uri else "?"
+    location = f"{redirect_uri}{sep}code={code}"
+    if state:
+        location += f"&state={state}"
+    return HTMLResponse(f'<script>window.location="{location}"</script>')
+
+
+async def handle_authorize(request: Request):
+    params = request.query_params
+    redirect_uri = params.get("redirect_uri")
+    state = params.get("state", "")
+    if not redirect_uri:
+        return JSONResponse({"error": "missing redirect_uri"}, status_code=400)
+
+    if not AUTH_TOKEN:
+        # No secret configured — behave as before (fine for trusted localhost-only use).
+        return _redirect_with_code(redirect_uri, state)
+
+    submitted = params.get("token")
+    if submitted is None:
+        return HTMLResponse(_authorize_form_html(redirect_uri, state))
+    if submitted != AUTH_TOKEN:
+        return HTMLResponse(_authorize_form_html(redirect_uri, state, error="Wrong token"), status_code=401)
+    return _redirect_with_code(redirect_uri, state)
+
+
+async def handle_register(request: Request):
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+
+    redirect_uris = body.get("redirect_uris", [])
+    return JSONResponse(
+        {
+            "client_id": "claude-ai-client",
+            "client_secret": "not-used",
+            "client_id_issued_at": int(time.time()),
+            "client_secret_expires_at": 0,
+            "redirect_uris": redirect_uris,
+            "grant_types": body.get("grant_types", ["authorization_code"]),
+            "response_types": body.get("response_types", ["code"]),
+            "token_endpoint_auth_method": "none",
+            "application_type": "web",
+        },
+        status_code=201,
+    )
+
+
+async def handle_token(request: Request):
+    if not AUTH_TOKEN:
+        return JSONResponse({"access_token": "ida-mcp-open-access", "token_type": "bearer", "expires_in": 86400})
+
+    try:
+        form = await request.form()
+        code = form.get("code")
+    except Exception:
+        code = None
+    if not code:
+        try:
+            body = await request.json()
+            code = body.get("code")
+        except Exception:
+            code = None
+
+    expiry = AUTH_CODES.pop(code, None) if code else None
+    if not expiry or expiry < time.time():
+        return JSONResponse({"error": "invalid_grant", "error_description": "missing/expired/used code"}, status_code=400)
+
+    return JSONResponse({"access_token": AUTH_TOKEN, "token_type": "bearer", "expires_in": 86400})
+
+
+# ─────────────────────────────────────────────
+#  App
+# ─────────────────────────────────────────────
+
+
+def create_app():
+    return Starlette(
+        routes=[
+            Route("/health", handle_health),
+            Route("/instances", handle_list_instances),
+            Route("/instances/register", handle_register_instance, methods=["POST"]),
+            Route("/instances/unregister", handle_unregister_instance, methods=["POST"]),
+            Route("/.well-known/oauth-protected-resource", handle_oauth_protected_resource),
+            Route("/.well-known/oauth-protected-resource/sse", handle_oauth_protected_resource),
+            Route("/.well-known/oauth-authorization-server", handle_oauth_authorization_server),
+            Route("/register", handle_register, methods=["POST"]),
+            Route("/authorize", handle_authorize, methods=["GET"]),
+            Route("/token", handle_token, methods=["POST"]),
+            Route("/sse", handle_mcp_post, methods=["GET", "POST", "HEAD"]),
+        ]
+    )
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    print(f"[IDA MCP] Dynamic multi-instance mode. MCP endpoint: http://0.0.0.0:{SSE_PORT}/sse")
+    print(f"[IDA MCP] Waiting for IDA plugin(s) to self-register on /instances/register ...")
+    if not AUTH_TOKEN:
+        print("[IDA MCP] WARNING: IDA_MCP_TOKEN is not set — server is open to anyone who can reach it.")
+        print("[IDA MCP]          Set IDA_MCP_TOKEN before exposing this via ngrok/the internet.")
+    if NGROK_URL:
+        print(f"[IDA MCP] NGROK_URL: {NGROK_URL}")
+    uvicorn.run(create_app(), host="0.0.0.0", port=SSE_PORT)
diff --git a/ida_mcp_plugin.py b/ida_mcp_plugin.py
new file mode 100644
index 0000000..d06fa48
--- /dev/null
+++ b/ida_mcp_plugin.py
@@ -0,0 +1,1506 @@
+"""
+IDA MCP Plugin.
+
+Each instance finds a free TCP port on its own, starts an HTTP server on it,
+and every 10s sends a heartbeat registration to the MCP server (IDA_MCP_REGISTRY_URL,
+defaults to http://127.0.0.1:8888). Open as many IDA instances as you want -- each
+registers itself under its own id (filename_pid); the MCP server sees all of them
+through ida_list_instances. No hardcoded ports, no manual setup per instance.
+
+Env vars:
+  IDA_MCP_PORT           force a specific port (default: auto-pick)
+  IDA_MCP_BASE_PORT      where auto-pick starts from (default 7777)
+  IDA_MCP_HOST           bind address (default 127.0.0.1; 0.0.0.0 for remote/SSH)
+  IDA_MCP_ADVERTISE_HOST address to report back to the MCP server (defaults to
+                         IDA_MCP_HOST unless that's 0.0.0.0, then 127.0.0.1 --
+                         set explicitly for a remote machine)
+  IDA_MCP_REGISTRY_URL   MCP server base URL for self-registration (default
+                         http://127.0.0.1:8888; empty disables registration)
+  IDA_MCP_TOKEN          shared secret, checked on incoming requests here and sent
+                         as Authorization: Bearer when registering with the MCP
+                         server -- set the same value on both sides
+
+See README.md for the full endpoint list.
+"""
+
+import json
+import os
+import socket
+import socketserver
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from urllib.parse import parse_qs, urlparse
+from urllib.request import Request as URLRequest, urlopen
+
+import idaapi
+import idautils
+import idc
+
+# ─────────────────────────────────────────────
+#  Config / self-registration
+# ─────────────────────────────────────────────
+
+BASE_PORT = int(os.environ.get("IDA_MCP_BASE_PORT", 7777))
+BIND_HOST = os.environ.get("IDA_MCP_HOST", "127.0.0.1")
+ADVERTISE_HOST = os.environ.get(
+    "IDA_MCP_ADVERTISE_HOST", BIND_HOST if BIND_HOST != "0.0.0.0" else "127.0.0.1"
+)
+REGISTRY_URL = os.environ.get("IDA_MCP_REGISTRY_URL", "http://127.0.0.1:8888")
+TOKEN = os.environ.get("IDA_MCP_TOKEN")
+HEARTBEAT_INTERVAL = 10
+
+PORT = None  # set in _start()
+INSTANCE_ID = None  # set in _start()
+
+
+def _find_free_port(start: int, host: str, tries: int = 50) -> int:
+    for p in range(start, start + tries):
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        try:
+            s.bind((host if host != "0.0.0.0" else "", p))
+            return p
+        except OSError:
+            continue
+        finally:
+            s.close()
+    raise RuntimeError(f"No free port found in range {start}-{start + tries}")
+
+
+def _make_instance_id() -> str:
+    try:
+        base = os.path.splitext(os.path.basename(idc.get_input_file_path() or "ida"))[0]
+        base = base or "ida"
+    except Exception:
+        base = "ida"
+    return f"{base}_{os.getpid()}"
+
+
+def _registry_call(path: str, payload: dict):
+    if not REGISTRY_URL:
+        return
+    try:
+        data = json.dumps(payload).encode("utf-8")
+        req = URLRequest(
+            f"{REGISTRY_URL}{path}",
+            data=data,
+            method="POST",
+            headers={"Content-Type": "application/json"},
+        )
+        if TOKEN:
+            req.add_header("Authorization", f"Bearer {TOKEN}")
+        urlopen(req, timeout=5).read()
+    except Exception as e:
+        print(f"[IDA MCP:{PORT}] registry {path} failed: {e}")
+
+
+def _heartbeat_loop(stop_event: threading.Event):
+    while not stop_event.is_set():
+        try:
+            path = idc.get_input_file_path()
+        except Exception:
+            path = "unknown"
+        _registry_call(
+            "/instances/register",
+            {
+                "id": INSTANCE_ID,
+                "host": ADVERTISE_HOST,
+                "port": PORT,
+                "file": path,
+                "pid": os.getpid(),
+            },
+        )
+        stop_event.wait(HEARTBEAT_INTERVAL)
+
+
+# ─────────────────────────────────────────────
+#  Auth
+# ─────────────────────────────────────────────
+
+
+def _check_auth(handler) -> bool:
+    if not TOKEN:
+        return True
+    return handler.headers.get("Authorization", "") == f"Bearer {TOKEN}"
+
+
+# ─────────────────────────────────────────────
+#  Name cache (str_* and other non-standard symbols)
+# ─────────────────────────────────────────────
+
+_name_cache = {}  # { name_str -> ea }
+_name_cache_built = False
+_name_cache_lock = threading.Lock()
+
+
+def _build_name_cache():
+    """Builds the full name cache via idautils.Names(). Called once."""
+    global _name_cache, _name_cache_built
+    cache = {}
+    for ea, name in idautils.Names():
+        cache[name] = ea
+    with _name_cache_lock:
+        _name_cache = cache
+        _name_cache_built = True
+    print(f"[IDA MCP:{PORT}] Name cache built: {len(cache)} entries")
+
+
+def _ensure_cache():
+    global _name_cache_built
+    if not _name_cache_built:
+        _build_name_cache()
+
+
+def _lookup_name(name: str) -> int:
+    if name.startswith("0x") or name.startswith("0X"):
+        return int(name, 16)
+
+    ea = idc.get_name_ea_simple(name)
+    if ea != idc.BADADDR:
+        return ea
+
+    # The decompiler shows str_ instead of .str. -- try both forms
+    alt_name = None
+    if name.startswith("str_"):
+        alt_name = ".str." + name[4:]
+    elif name.startswith("str"):
+        alt_name = ".str" + name[3:]
+
+    if alt_name:
+        ea = idc.get_name_ea_simple(alt_name)
+        if ea != idc.BADADDR:
+            return ea
+
+    _ensure_cache()
+    with _name_cache_lock:
+        ea = _name_cache.get(name, idc.BADADDR)
+        if ea == idc.BADADDR and alt_name:
+            ea = _name_cache.get(alt_name, idc.BADADDR)
+        return ea
+
+
+# ─────────────────────────────────────────────
+#  Thread safety
+# ─────────────────────────────────────────────
+
+
+def run_in_main_thread(fn):
+    """Runs fn on the IDA main thread synchronously, returns its result."""
+    result = [None]
+    error = [None]
+    event = threading.Event()
+
+    def callback():
+        try:
+            result[0] = fn()
+        except Exception as e:
+            error[0] = e
+        finally:
+            event.set()
+        return 0
+
+    idaapi.execute_sync(callback, idaapi.MFF_FAST)
+    if not event.wait(timeout=90):
+        raise TimeoutError(
+            "IDA main thread did not respond in time (busy with another operation?)"
+        )
+
+    if error[0]:
+        raise error[0]
+    return result[0]
+
+
+# ─────────────────────────────────────────────
+#  Resolve helper
+# ─────────────────────────────────────────────
+
+
+class ApiError(ValueError):
+    """ValueError with a machine-readable code, so callers can branch on it
+    instead of parsing the error text."""
+
+    def __init__(self, message: str, code: str = "bad_request"):
+        super().__init__(message)
+        self.code = code
+
+
+def _resolve_inner(name: str) -> int:
+    if name.startswith("0x") or name.startswith("0X"):
+        return int(name, 16)
+    ea = idc.get_name_ea_simple(name)
+    if ea == idc.BADADDR:
+        raise ApiError(f"Symbol not found: {name}", code="not_found")
+    return ea
+
+
+def _require(body: dict, fields):
+    missing = [f for f in fields if f not in body or body[f] is None]
+    if missing:
+        raise ApiError(f"Missing required field(s): {', '.join(missing)}", code="bad_request")
+
+
+# ─────────────────────────────────────────────
+#  IDA operations
+# ─────────────────────────────────────────────
+
+
+def _ping():
+    def _():
+        try:
+            path = idc.get_input_file_path()
+        except Exception:
+            path = "unknown"
+        try:
+            import ida_hexrays
+
+            hexrays_ok = ida_hexrays.init_hexrays_plugin()
+        except Exception:
+            hexrays_ok = False
+        try:
+            analysis_complete = idaapi.auto_is_ok()
+        except Exception:
+            analysis_complete = None
+        return {
+            "status": "ok",
+            "id": INSTANCE_ID,
+            "host": ADVERTISE_HOST,
+            "port": PORT,
+            "file": path,
+            "hexrays_available": bool(hexrays_ok),
+            "analysis_complete": analysis_complete,
+            "name_cache": len(_name_cache) if _name_cache_built else "not built",
+            "registry_url": REGISTRY_URL or None,
+        }
+
+    return run_in_main_thread(_)
+
+
+def _list_functions(filter_str=None, limit=200, offset=0):
+    def _():
+        import ida_funcs
+
+        funcs = []
+        for ea in idautils.Functions():
+            name = idc.get_func_name(ea)
+            if filter_str and filter_str.lower() not in name.lower():
+                continue
+            f = ida_funcs.get_func(ea)
+            if f:
+                funcs.append({"ea": hex(f.start_ea), "name": name, "size": f.size()})
+        total = len(funcs)
+        return {
+            "total": total,
+            "offset": offset,
+            "limit": limit,
+            "functions": funcs[offset : offset + limit],
+        }
+
+    return run_in_main_thread(_)
+
+
+def _search_functions(q: str, limit=100):
+    def _():
+        results = []
+        q_lower = q.lower()
+        for ea in idautils.Functions():
+            name = idc.get_func_name(ea)
+            if q_lower in name.lower():
+                results.append({"ea": hex(ea), "name": name})
+                if len(results) >= limit:
+                    break
+        return {"count": len(results), "functions": results}
+
+    return run_in_main_thread(_)
+
+
+def _search_strings(q: str, limit=50):
+    def _():
+        results = []
+        q_lower = q.lower()
+        for s in idautils.Strings():
+            try:
+                val = str(s)
+                if q_lower in val.lower():
+                    results.append({"ea": hex(s.ea), "value": val})
+                    if len(results) >= limit:
+                        break
+            except Exception:
+                pass
+        return {"count": len(results), "strings": results}
+
+    return run_in_main_thread(_)
+
+
+def _decompile_one(target: str):
+    """Returns a dict with the code, or an explicit 'error' field (never embedded in the code text)."""
+    import ida_funcs
+    import ida_hexrays
+
+    ea = _resolve_inner(target)
+    name = idc.get_func_name(ea)
+    f = ida_funcs.get_func(ea)
+    try:
+        cfunc = ida_hexrays.decompile(ea)
+    except Exception as e:
+        return {"target": target, "ea": hex(ea), "name": name, "error": f"decompile exception: {e}"}
+    if not cfunc:
+        return {
+            "target": target,
+            "ea": hex(ea),
+            "name": name,
+            "error": "decompile failed (no Hex-Rays for this arch, or not a function?)",
+        }
+    return {
+        "target": target,
+        "ea": hex(ea),
+        "name": name,
+        "size": f.size() if f else 0,
+        "code": str(cfunc),
+    }
+
+
+def _decompile(target: str):
+    def _():
+        return _decompile_one(target)
+
+    return run_in_main_thread(_)
+
+
+def _decompile_many(targets):
+    def _():
+        results = []
+        for t in targets:
+            try:
+                results.append(_decompile_one(t))
+            except Exception as e:
+                results.append({"target": t, "error": str(e)})
+        return {"count": len(results), "results": results}
+
+    return run_in_main_thread(_)
+
+
+def _get_disasm(target: str, count: int = 30, offset: int = 0):
+    def _():
+        import ida_lines
+        import ida_funcs
+
+        ea = _resolve_inner(target)
+        f = ida_funcs.get_func(ea)
+        end = f.end_ea if f else None
+
+        # Known instruction count up front, instead of guessing from what's returned.
+        total = len(list(idautils.FuncItems(ea))) if f else None
+
+        lines = []
+        cur = ea
+        n = 0
+        skipped = 0
+        while cur != idc.BADADDR:
+            if end is not None and cur >= end:
+                break
+            if skipped < offset:
+                skipped += 1
+                cur = idc.next_head(cur, end if end is not None else cur + 0x2000)
+                continue
+            if n >= count:
+                break
+            raw = idc.generate_disasm_line(cur, 0) or ""
+            lines.append({"ea": hex(cur), "text": ida_lines.tag_remove(raw)})
+            cur = idc.next_head(cur, end if end is not None else cur + 0x2000)
+            n += 1
+
+        truncated = cur != idc.BADADDR and (end is None or cur < end)
+        return {
+            "ea": hex(ea),
+            "func_end": hex(end) if end is not None else None,
+            "total_instructions": total,
+            "offset": offset,
+            "returned": len(lines),
+            "truncated": truncated,
+            "lines": lines,
+        }
+
+    return run_in_main_thread(_)
+
+
+def _xrefs_to(target: str):
+    def _():
+        ea = _resolve_inner(target)
+        results = []
+        for xref in idautils.XrefsTo(ea, 0):
+            from_ea = xref.frm
+            func = idaapi.get_func(from_ea)
+            results.append(
+                {
+                    "from_ea": hex(from_ea),
+                    "from_func": idc.get_func_name(from_ea) or "",
+                    "func_ea": hex(func.start_ea) if func else hex(from_ea),
+                }
+            )
+        return results
+
+    return run_in_main_thread(_)
+
+
+def _xrefs_to_many(targets):
+    def _():
+        result = {}
+        for t in targets:
+            try:
+                ea = _resolve_inner(t)
+                entries = []
+                for xref in idautils.XrefsTo(ea, 0):
+                    from_ea = xref.frm
+                    func = idaapi.get_func(from_ea)
+                    entries.append(
+                        {
+                            "from_ea": hex(from_ea),
+                            "from_func": idc.get_func_name(from_ea) or "",
+                            "func_ea": hex(func.start_ea) if func else hex(from_ea),
+                        }
+                    )
+                result[t] = entries
+            except Exception as e:
+                result[t] = {"error": str(e)}
+        return result
+
+    return run_in_main_thread(_)
+
+
+def _xrefs_from(target: str):
+    def _():
+        ea = _resolve_inner(target)
+        return [
+            {"to_ea": hex(r), "to_func": idc.get_func_name(r)}
+            for r in idautils.CodeRefsFrom(ea, 0)
+        ]
+
+    return run_in_main_thread(_)
+
+
+def _get_imports(filter_str=None, limit=500):
+    def _():
+        # IDA 9.x moved these from idaapi to ida_nalt and renamed _count -> _qty;
+        # fall back to the old idaapi names for older IDA versions.
+        try:
+            import ida_nalt as _imp
+
+            qty = _imp.get_import_module_qty()
+        except (ImportError, AttributeError):
+            _imp = idaapi
+            qty = idaapi.get_import_module_count()
+
+        imports = []
+        for i in range(qty):
+            mod = _imp.get_import_module_name(i)
+
+            def cb(ea, name, ord, mod=mod):
+                if not filter_str or filter_str.lower() in (name or "").lower():
+                    imports.append(
+                        {"module": mod, "name": name or f"ord_{ord}", "ea": hex(ea)}
+                    )
+                return True
+
+            _imp.enum_import_names(i, cb)
+        return imports[:limit]
+
+    return run_in_main_thread(_)
+
+
+def _get_segments():
+    def _():
+        import ida_segment
+
+        segs = []
+        for ea in idautils.Segments():
+            seg = ida_segment.getseg(ea)
+            segs.append(
+                {
+                    "name": idc.get_segm_name(ea),
+                    "start": hex(seg.start_ea),
+                    "end": hex(seg.end_ea),
+                    "size": seg.end_ea - seg.start_ea,
+                    "perm": {
+                        "r": bool(seg.perm & ida_segment.SEGPERM_READ),
+                        "w": bool(seg.perm & ida_segment.SEGPERM_WRITE),
+                        "x": bool(seg.perm & ida_segment.SEGPERM_EXEC),
+                    },
+                }
+            )
+        return segs
+
+    return run_in_main_thread(_)
+
+
+def _get_entry_points():
+    def _():
+        import ida_entry
+
+        result = []
+        for i in range(ida_entry.get_entry_qty()):
+            ordinal = ida_entry.get_entry_ordinal(i)
+            ea = ida_entry.get_entry(ordinal)
+            name = ida_entry.get_entry_name(ordinal)
+            result.append({"ordinal": ordinal, "ea": hex(ea), "name": name})
+        return result
+
+    return run_in_main_thread(_)
+
+
+def _get_prototype(target: str):
+    def _():
+        ea = _resolve_inner(target)
+        return {"ea": hex(ea), "name": idc.get_func_name(ea), "prototype": idc.get_type(ea) or ""}
+
+    return run_in_main_thread(_)
+
+
+def _set_prototype(target: str, prototype: str):
+    def _():
+        ea = _resolve_inner(target)
+        decl = prototype if prototype.rstrip().endswith(";") else prototype + ";"
+        ok = idc.set_type(ea, decl)
+        if not ok:
+            raise ValueError(f"IDA rejected prototype: {prototype}")
+        return {"ok": True, "ea": hex(ea), "prototype": idc.get_type(ea)}
+
+    return run_in_main_thread(_)
+
+
+def _set_lvar_type(function: str, lvar_name: str, type_str: str):
+    """Experimental: the Hex-Rays API for changing a local variable's type can
+    differ slightly across IDA versions. Errors are surfaced as-is so it's
+    clear what to fix."""
+
+    def _():
+        import ida_hexrays
+        import ida_typeinf
+
+        ea = _resolve_inner(function)
+        cfunc = ida_hexrays.decompile(ea)
+        if not cfunc:
+            raise ValueError("Cannot decompile function")
+
+        tif = ida_typeinf.tinfo_t()
+        decl = type_str.strip()
+        parseable = decl if decl.endswith(";") else decl + " x;"
+        if not ida_typeinf.parse_decl(tif, None, parseable, ida_typeinf.PT_SIL) and not tif.get_named_type(
+            None, type_str
+        ):
+            raise ValueError(f"Cannot parse type: {type_str}")
+
+        for lvar in cfunc.get_lvars():
+            if lvar.name == lvar_name:
+                if not lvar.set_lvar_type(tif):
+                    raise ValueError(
+                        "set_lvar_type() rejected by Hex-Rays (incompatible type or read-only var)"
+                    )
+                cfunc.save_user_lvar_settings()
+                return {"ok": True, "function": idc.get_func_name(ea), "lvar": lvar_name, "type": type_str}
+        raise ValueError(f"Local variable '{lvar_name}' not found")
+
+    return run_in_main_thread(_)
+
+
+def _rename_function(target: str, new_name: str):
+    def _():
+        ea = _resolve_inner(target)
+        old_name = idc.get_func_name(ea)
+        ok = idc.set_name(ea, new_name, idc.SN_NOWARN)
+        if not ok:
+            raise ValueError(f"IDA rejected name '{new_name}' (duplicate or invalid?)")
+        with _name_cache_lock:
+            if old_name in _name_cache:
+                del _name_cache[old_name]
+            _name_cache[new_name] = ea
+        return {"ok": True, "ea": hex(ea), "old_name": old_name, "new_name": new_name}
+
+    return run_in_main_thread(_)
+
+
+def _set_comment(target: str, comment: str):
+    def _():
+        ea = _resolve_inner(target)
+        idc.set_func_cmt(ea, comment, 0)
+        return {"ok": True}
+
+    return run_in_main_thread(_)
+
+
+def _rename_local(function: str, old_name: str, new_name: str):
+    def _():
+        import ida_hexrays
+
+        ea = _resolve_inner(function)
+        cfunc = ida_hexrays.decompile(ea)
+        if not cfunc:
+            raise ValueError("Cannot decompile function")
+        for lvar in cfunc.get_lvars():
+            if lvar.name == old_name:
+                lvar.name = new_name
+                cfunc.save_user_lvar_settings()
+                return {"ok": True}
+        raise ValueError(f"Local variable '{old_name}' not found")
+
+    return run_in_main_thread(_)
+
+
+def _get_name_value(name: str):
+    def _():
+        ea = _lookup_name(name)
+
+        if ea == idc.BADADDR:
+            raise ValueError(f"Not found: {name}")
+
+        s = idc.get_strlit_contents(ea, -1, idc.STRTYPE_C)
+        if s is not None:
+            return {
+                "ea": hex(ea),
+                "type": "string",
+                "value": s.decode("utf-8", errors="replace"),
+            }
+
+        val = idc.get_wide_dword(ea)
+        return {"ea": hex(ea), "type": "dword", "value": hex(val)}
+
+    return run_in_main_thread(_)
+
+
+def _rebuild_name_cache():
+    def _():
+        global _name_cache_built
+        _name_cache_built = False
+        _build_name_cache()
+        return {"ok": True, "entries": len(_name_cache)}
+
+    return run_in_main_thread(_)
+
+
+def _reload_plugin_code():
+    """Re-reads this .py file from disk and swaps in the new handler functions
+    without restarting IDA. This works because do_GET/do_POST look up
+    _decompile/_get_imports/etc. by name in the module's global namespace on
+    every request -- after reload, that lookup finds the new code.
+
+    Limitation: PORT/INSTANCE_ID are restored manually (reload would otherwise
+    reset them), and the heartbeat thread/HTTPServer keep running with their
+    old code until an actual IDA restart, since they were started with a
+    direct function reference rather than a by-name lookup."""
+
+    def _():
+        import importlib
+        import sys
+
+        modname = __name__
+        mod = sys.modules.get(modname)
+        if mod is None:
+            raise ValueError(f"Cannot find own module in sys.modules as '{modname}'")
+
+        saved_port, saved_id = PORT, INSTANCE_ID
+        try:
+            importlib.reload(mod)
+        except Exception:
+            # IDA-loaded plugin modules often have no usable importlib spec
+            # (loaded via IDA's own mechanism, not a normal import) — fall back
+            # to exec'ing the file straight into the module's own namespace.
+            try:
+                with open(mod.__file__, "r", encoding="utf-8") as f:
+                    src = f.read()
+                exec(compile(src, mod.__file__, "exec"), mod.__dict__)
+            except Exception as e:
+                raise ValueError(f"Reload failed (syntax error in the file?): {e}")
+
+        mod.PORT = saved_port
+        mod.INSTANCE_ID = saved_id
+        mod._name_cache_built = False
+        threading.Thread(target=lambda: run_in_main_thread(mod._build_name_cache), daemon=True).start()
+
+        return {
+            "ok": True,
+            "port": mod.PORT,
+            "id": mod.INSTANCE_ID,
+            "note": "Endpoint handlers reloaded. Heartbeat thread keeps its old code until a full IDA restart.",
+        }
+
+    return run_in_main_thread(_)
+
+
+# ─────────────────────────────────────────────
+#  Notes (netnode-backed -- persist across sessions and tools)
+# ─────────────────────────────────────────────
+
+_NOTES_NODE_NAME = "$ ida_mcp_notes"
+
+
+def _notes_node():
+    import ida_netnode
+
+    return ida_netnode.netnode(_NOTES_NODE_NAME, 0, True)
+
+
+def _get_notes(target: str):
+    def _():
+        ea = _resolve_inner(target)
+        val = _notes_node().supval(ea)
+        text = val.decode("utf-8", errors="replace") if val else ""
+        return {"ea": hex(ea), "note": text}
+
+    return run_in_main_thread(_)
+
+
+def _set_notes(target: str, note: str):
+    def _():
+        ea = _resolve_inner(target)
+        nn = _notes_node()
+        if note:
+            nn.supset(ea, note.encode("utf-8"))
+        else:
+            nn.supdel(ea)
+        return {"ok": True, "ea": hex(ea)}
+
+    return run_in_main_thread(_)
+
+
+# ─────────────────────────────────────────────
+#  Bytes / binary search
+# ─────────────────────────────────────────────
+
+
+def _get_bytes(target: str, size: int = 64):
+    def _():
+        ea = _resolve_inner(target)
+        data = idc.get_bytes(ea, size)
+        if data is None:
+            raise ValueError(f"Cannot read {size} bytes at {hex(ea)} (unmapped?)")
+        return {"ea": hex(ea), "size": size, "hex": data.hex()}
+
+    return run_in_main_thread(_)
+
+
+def _find_pattern(pattern: str, max_results: int = 50):
+    """Searches for an IDA-style hex pattern ("48 8B ?? ??") across the whole image.
+    Experimental: the bin_search API has changed across IDA versions; falls
+    back to the legacy idc.find_binary."""
+
+    def _():
+        results = []
+        modern_ok = False
+        modern_err = None
+        try:
+            import ida_bytes
+
+            patterns = ida_bytes.compiled_binpat_vec_t()
+            if not ida_bytes.parse_binpat_str(patterns, 0, pattern, 16):
+                modern_err = "parse_binpat_str() returned False (bad pattern syntax?)"
+            else:
+                modern_ok = True
+                ea = 0
+                while len(results) < max_results:
+                    found = ida_bytes.bin_search(ea, idaapi.BADADDR, patterns, ida_bytes.BIN_SEARCH_FORWARD)
+                    found_ea = found[0] if isinstance(found, tuple) else found
+                    if found_ea is None or found_ea == idaapi.BADADDR:
+                        break
+                    results.append(hex(found_ea))
+                    ea = found_ea + 1
+        except Exception as e:
+            modern_ok = False
+            modern_err = f"{type(e).__name__}: {e}"
+            results = []
+
+        if not modern_ok:
+            try:
+                ea = idc.find_binary(0, idc.SEARCH_DOWN, pattern)
+                while ea != idc.BADADDR and len(results) < max_results:
+                    results.append(hex(ea))
+                    ea = idc.find_binary(ea + 1, idc.SEARCH_DOWN, pattern)
+            except Exception as e:
+                raise ValueError(
+                    f"bin_search path failed ({modern_err}); legacy find_binary also failed: {type(e).__name__}: {e}"
+                )
+
+        return {"pattern": pattern, "count": len(results), "matches": results, "used_legacy_find_binary": not modern_ok}
+
+    return run_in_main_thread(_)
+
+
+def _patch_bytes(target: str, hex_bytes: str):
+    def _():
+        import ida_bytes
+
+        ea = _resolve_inner(target)
+        data = bytes.fromhex(hex_bytes.replace(" ", ""))
+        for i, b in enumerate(data):
+            ida_bytes.patch_byte(ea + i, b)
+        return {"ok": True, "ea": hex(ea), "patched_bytes": len(data)}
+
+    return run_in_main_thread(_)
+
+
+def _make_data(target: str, data_type: str, struct_name: str = None):
+    """Experimental: some struct-related idc functions are legacy and may have moved in IDA 9.x."""
+
+    def _():
+        import ida_bytes
+
+        ea = _resolve_inner(target)
+        sizes = {"byte": (ida_bytes.FF_BYTE, 1), "word": (ida_bytes.FF_WORD, 2), "dword": (ida_bytes.FF_DWORD, 4), "qword": (ida_bytes.FF_QWORD, 8)}
+        if data_type in sizes:
+            flag, size = sizes[data_type]
+            ok = ida_bytes.create_data(ea, flag, size, idaapi.BADADDR)
+        elif data_type == "struct":
+            if not struct_name:
+                raise ValueError("struct_name is required for data_type='struct'")
+            tid = idc.get_struc_id(struct_name)
+            if tid == idc.BADADDR:
+                raise ValueError(f"Struct not found: {struct_name}")
+            size = idc.get_struc_size(tid)
+            ok = ida_bytes.create_struct(ea, size, tid)
+        else:
+            raise ValueError(f"Unknown data_type: {data_type} (use byte/word/dword/qword/struct)")
+        if not ok:
+            raise ValueError("Failed to create data (already defined at this address, or bad location?)")
+        return {"ok": True, "ea": hex(ea), "type": data_type}
+
+    return run_in_main_thread(_)
+
+
+# ─────────────────────────────────────────────
+#  Comments at a specific address (set_comment is whole-function)
+# ─────────────────────────────────────────────
+
+
+def _get_line_comment(target: str, repeatable: bool = False):
+    def _():
+        ea = _resolve_inner(target)
+        return {"ea": hex(ea), "comment": idc.get_cmt(ea, repeatable) or ""}
+
+    return run_in_main_thread(_)
+
+
+def _set_line_comment(target: str, comment: str, repeatable: bool = False):
+    def _():
+        ea = _resolve_inner(target)
+        idc.set_cmt(ea, comment, repeatable)
+        return {"ok": True, "ea": hex(ea)}
+
+    return run_in_main_thread(_)
+
+
+# ─────────────────────────────────────────────
+#  Triage: top functions, call graph
+# ─────────────────────────────────────────────
+
+
+def _top_functions(by: str = "xrefs", limit: int = 20):
+    def _():
+        import ida_funcs
+
+        stats = []
+        for ea in idautils.Functions():
+            f = ida_funcs.get_func(ea)
+            if not f:
+                continue
+            if by == "size":
+                score = f.size()
+            else:
+                score = sum(1 for _ in idautils.XrefsTo(ea, 0))
+            stats.append({"ea": hex(ea), "name": idc.get_func_name(ea), by: score})
+        stats.sort(key=lambda x: x[by], reverse=True)
+        return stats[:limit]
+
+    return run_in_main_thread(_)
+
+
+def _call_graph(target: str, depth: int = 2, direction: str = "callees", limit_per_node: int = 30):
+    def _():
+        import ida_funcs
+
+        start = _resolve_inner(target)
+
+        def neighbors(ea):
+            if direction == "callers":
+                found = set()
+                for x in idautils.XrefsTo(ea, 0):
+                    func = idaapi.get_func(x.frm)
+                    if func:
+                        found.add(func.start_ea)
+                return list(found)[:limit_per_node]
+            f = ida_funcs.get_func(ea)
+            if not f:
+                return []
+            found = set()
+            for head in idautils.FuncItems(ea):
+                for r in idautils.CodeRefsFrom(head, 0):
+                    func = idaapi.get_func(r)
+                    if func and func.start_ea != ea:
+                        found.add(func.start_ea)
+            return list(found)[:limit_per_node]
+
+        nodes = {start: 0}
+        edges = []
+        frontier = [start]
+        for d in range(depth):
+            next_frontier = []
+            for node in frontier:
+                for n in neighbors(node):
+                    edges.append({"from": hex(node), "to": hex(n)})
+                    if n not in nodes:
+                        nodes[n] = d + 1
+                        next_frontier.append(n)
+            frontier = next_frontier
+
+        return {
+            "direction": direction,
+            "nodes": [{"ea": hex(ea), "name": idc.get_func_name(ea), "depth": d} for ea, d in nodes.items()],
+            "edges": edges,
+        }
+
+    return run_in_main_thread(_)
+
+
+# ─────────────────────────────────────────────
+#  Functions: create/undefine
+# ─────────────────────────────────────────────
+
+
+def _make_function(target: str):
+    def _():
+        import ida_funcs
+
+        ea = _resolve_inner(target)
+        if not ida_funcs.add_func(ea):
+            raise ValueError(f"add_func failed at {hex(ea)} (already a function, or not valid code?)")
+        return {"ok": True, "ea": hex(ea), "name": idc.get_func_name(ea)}
+
+    return run_in_main_thread(_)
+
+
+def _undefine(target: str):
+    def _():
+        import ida_funcs
+
+        ea = _resolve_inner(target)
+        f = idaapi.get_func(ea)
+        if f:
+            ida_funcs.del_func(f.start_ea)
+        idc.del_items(ea, idc.DELIT_SIMPLE, 1)
+        return {"ok": True, "ea": hex(ea)}
+
+    return run_in_main_thread(_)
+
+
+# ─────────────────────────────────────────────
+#  Local types (structs/enums)
+# ─────────────────────────────────────────────
+
+
+def _list_local_types(limit: int = 200):
+    def _():
+        import ida_typeinf
+
+        til = ida_typeinf.get_idati()
+
+        qty = None
+        for getter in (
+            lambda: ida_typeinf.get_ordinal_qty(til),
+            lambda: til.get_ordinal_qty(),
+            lambda: ida_typeinf.get_ordinal_count(til),
+        ):
+            try:
+                qty = getter()
+                break
+            except Exception:
+                continue
+        if qty is None:
+            raise ValueError("Cannot determine local-types count on this IDA version (API changed)")
+
+        results = []
+        for ordinal in range(1, qty + 1):
+            tif = ida_typeinf.tinfo_t()
+            if tif.get_numbered_type(til, ordinal):
+                results.append(
+                    {
+                        "ordinal": ordinal,
+                        "name": tif.get_type_name() or "",
+                        "size": tif.get_size() if tif.get_size() != idaapi.BADSIZE else None,
+                        "is_struct": tif.is_struct(),
+                        "is_enum": tif.is_enum(),
+                    }
+                )
+            if len(results) >= limit:
+                break
+        return results
+
+    return run_in_main_thread(_)
+
+
+def _create_struct(name: str, fields):
+    """fields: [{"name": "field1", "type": "int"}, ...] -- adds a struct to Local Types."""
+
+    def _():
+        decl = f"struct {name} {{\n"
+        for f in fields:
+            decl += f"  {f['type']} {f['name']};\n"
+        decl += "};"
+        errors = idc.parse_decls(decl, False)
+        if errors != 0:
+            raise ValueError(f"parse_decls reported {errors} error(s) for:\n{decl}")
+        return {"ok": True, "name": name, "decl": decl}
+
+    return run_in_main_thread(_)
+
+
+def _create_local_type(decl: str):
+    """Generalized version of _create_struct -- takes a raw C declaration
+    (struct/enum/union/typedef), not just a struct built from a field list."""
+
+    def _():
+        errors = idc.parse_decls(decl, False)
+        if errors != 0:
+            raise ValueError(f"parse_decls reported {errors} error(s) for:\n{decl}")
+        return {"ok": True, "decl": decl}
+
+    return run_in_main_thread(_)
+
+
+# ─────────────────────────────────────────────
+#  Objective-C: classes/methods by naming convention (+[Class sel:], -[Class sel:])
+# ─────────────────────────────────────────────
+
+_OBJC_METHOD_RE = None  # compiled lazily
+
+
+def _find_immediate(value: str, limit: int = 100):
+    """Searches all instructions for an immediate operand matching the given value."""
+
+    def _():
+        try:
+            target = int(value, 0)
+        except ValueError:
+            raise ValueError(f"Cannot parse value: {value} (use decimal or 0x-hex)")
+
+        results = []
+        for func_ea in idautils.Functions():
+            for ea in idautils.FuncItems(func_ea):
+                for n in range(6):
+                    try:
+                        optype = idc.get_operand_type(ea, n)
+                    except Exception:
+                        break
+                    if optype == idc.o_void:
+                        break
+                    if optype == idc.o_imm and idc.get_operand_value(ea, n) == target:
+                        results.append({"ea": hex(ea), "func": idc.get_func_name(ea), "operand": n})
+                        if len(results) >= limit:
+                            return {"value": hex(target), "count": len(results), "matches": results}
+        return {"value": hex(target), "count": len(results), "matches": results}
+
+    return run_in_main_thread(_)
+
+
+def _find_vtables(segment_filter: str = None, min_entries: int = 2, limit: int = 50):
+    """Heuristic, experimental: looks for runs of consecutive pointers in
+    non-executable segments where each one points to the start of a real
+    function -- the classic vtable signature. Can be slow on large binaries
+    and gives false positives; pass segment_filter (e.g. "__const"/"__data")
+    to narrow the scan."""
+
+    def _():
+        import ida_segment
+        import ida_funcs
+
+        ptr_size = 8 if idaapi.get_inf_structure().is_64bit() else 4
+        get_ptr = idc.get_qword if ptr_size == 8 else idc.get_wide_dword
+
+        def is_func_start(val):
+            if not val or val == idc.BADADDR:
+                return False
+            f = ida_funcs.get_func(val)
+            return bool(f and f.start_ea == val)
+
+        candidates = []
+        for seg_ea in idautils.Segments():
+            seg = ida_segment.getseg(seg_ea)
+            if seg.perm & ida_segment.SEGPERM_EXEC:
+                continue
+            name = idc.get_segm_name(seg_ea)
+            if segment_filter and segment_filter.lower() not in name.lower():
+                continue
+
+            ea = seg.start_ea
+            while ea + ptr_size <= seg.end_ea and len(candidates) < limit:
+                if not is_func_start(get_ptr(ea)):
+                    ea += ptr_size
+                    continue
+                run_start = ea
+                entries = []
+                cur = ea
+                while cur + ptr_size <= seg.end_ea:
+                    ptr = get_ptr(cur)
+                    if not is_func_start(ptr):
+                        break
+                    entries.append({"ea": hex(cur), "func_ea": hex(ptr), "func_name": idc.get_func_name(ptr)})
+                    cur += ptr_size
+                if len(entries) >= min_entries:
+                    candidates.append({"ea": hex(run_start), "segment": name, "entries": entries})
+                ea = cur if cur > run_start else run_start + ptr_size
+            if len(candidates) >= limit:
+                break
+
+        return {"count": len(candidates), "vtables": candidates}
+
+    return run_in_main_thread(_)
+
+
+def _get_objc_classes(filter_str: str = None, limit: int = 200):
+    def _():
+        import re
+
+        global _OBJC_METHOD_RE
+        if _OBJC_METHOD_RE is None:
+            _OBJC_METHOD_RE = re.compile(r"^([+-])\[([\w]+)(\([\w]+\))?\s+([^\]]+)\]$")
+
+        _ensure_cache()
+        classes = {}
+
+        # 1) class objects themselves, from runtime symbols
+        with _name_cache_lock:
+            names = list(_name_cache.items())
+        for name, ea in names:
+            if name.startswith("_OBJC_CLASS_$_"):
+                cls = name[len("_OBJC_CLASS_$_") :]
+                classes.setdefault(cls, {"name": cls, "ea": hex(ea), "methods": []})
+
+        # 2) methods, from functions named +[Class sel] / -[Class sel]
+        for ea in idautils.Functions():
+            fname = idc.get_func_name(ea)
+            m = _OBJC_METHOD_RE.match(fname)
+            if not m:
+                continue
+            kind, cls, category, selector = m.groups()
+            entry = classes.setdefault(cls, {"name": cls, "ea": None, "methods": []})
+            entry["methods"].append(
+                {
+                    "kind": "class" if kind == "+" else "instance",
+                    "selector": selector,
+                    "category": category.strip("()") if category else None,
+                    "ea": hex(ea),
+                }
+            )
+
+        result = list(classes.values())
+        if filter_str:
+            fl = filter_str.lower()
+            result = [c for c in result if fl in c["name"].lower()]
+        return {"count": len(result), "classes": result[:limit]}
+
+    return run_in_main_thread(_)
+
+
+# ─────────────────────────────────────────────
+#  HTTP handler
+# ─────────────────────────────────────────────
+
+HELP_ROUTES = [
+    {"method": "GET", "path": "/ping", "desc": "Instance status, Hex-Rays availability"},
+    {"method": "GET", "path": "/help", "desc": "This list"},
+    {"method": "GET", "path": "/functions?filter=&limit=&offset=", "desc": "List functions"},
+    {"method": "GET", "path": "/functions/search?q=&limit=", "desc": "Search functions by substring"},
+    {"method": "GET", "path": "/strings/search?q=&limit=", "desc": "Search strings"},
+    {"method": "GET", "path": "/name/<name_or_0xADDR>", "desc": "Value at a named symbol"},
+    {"method": "GET", "path": "/name_cache/rebuild", "desc": "Rebuild the name cache"},
+    {"method": "GET", "path": "/imports?filter=", "desc": "Import table"},
+    {"method": "GET", "path": "/segments", "desc": "Segments with permissions"},
+    {"method": "GET", "path": "/entry_points", "desc": "Entry points"},
+    {"method": "GET", "path": "/decompile/<target>", "desc": "Decompile a function"},
+    {"method": "POST", "path": "/decompile_batch", "desc": '{"targets": [...]} -- decompile in one call'},
+    {"method": "GET", "path": "/disasm/<target>?count=", "desc": "Disassemble a function"},
+    {"method": "GET", "path": "/xrefs_to/<target>", "desc": "Who references this address"},
+    {"method": "GET", "path": "/xrefs_from/<target>", "desc": "What this function references"},
+    {"method": "GET", "path": "/prototype/<target>", "desc": "Function prototype"},
+    {"method": "POST", "path": "/set_prototype", "desc": '{"target","prototype"}'},
+    {"method": "POST", "path": "/set_lvar_type", "desc": '{"function","lvar_name","type"} -- experimental'},
+    {"method": "GET", "path": "/notes/<target>", "desc": "Note stored at an address"},
+    {"method": "POST", "path": "/notes", "desc": '{"target","note"} -- note="" deletes it'},
+    {"method": "POST", "path": "/rename_function", "desc": '{"target","new_name"}'},
+    {"method": "POST", "path": "/rename_local", "desc": '{"function","old_name","new_name"}'},
+    {"method": "POST", "path": "/set_comment", "desc": '{"target","comment"}'},
+    {"method": "GET", "path": "/bytes/<target>?size=", "desc": "Raw bytes at an address (hex)"},
+    {"method": "GET", "path": "/find?pattern=&limit=", "desc": 'Search a hex pattern, e.g. "48 8B ?? ??" -- experimental'},
+    {"method": "POST", "path": "/patch_bytes", "desc": '{"target","hex_bytes"} -- writes to the binary!'},
+    {"method": "POST", "path": "/make_data", "desc": '{"target","data_type":"byte|word|dword|qword|struct","struct_name"} -- experimental'},
+    {"method": "GET", "path": "/line_comment/<target>?repeatable=", "desc": "Comment at a specific address (not the whole function)"},
+    {"method": "POST", "path": "/line_comment", "desc": '{"target","comment","repeatable"}'},
+    {"method": "GET", "path": "/top_functions?by=xrefs|size&limit=", "desc": "Triage: most-referenced/largest functions"},
+    {"method": "GET", "path": "/call_graph/<target>?depth=&direction=callees|callers", "desc": "Call graph N levels deep"},
+    {"method": "POST", "path": "/make_function", "desc": '{"target"} -- create a function at an address'},
+    {"method": "POST", "path": "/undefine", "desc": '{"target"} -- remove function/data definition at an address'},
+    {"method": "GET", "path": "/local_types?limit=", "desc": "List structs/enums from Local Types"},
+    {"method": "POST", "path": "/create_struct", "desc": '{"name","fields":[{"name","type"},...]}'},
+    {"method": "GET", "path": "/objc_classes?filter=&limit=", "desc": "Objective-C classes/methods by naming convention"},
+    {"method": "GET", "path": "/find_immediate?value=&limit=", "desc": "Search for an immediate operand with a given value"},
+    {"method": "GET", "path": "/find_vtables?segment=&min_entries=&limit=", "desc": "Heuristic vtable-like pointer array search -- experimental"},
+    {"method": "POST", "path": "/create_type", "desc": '{"decl"} -- raw C declaration for struct/enum/union/typedef in Local Types'},
+    {"method": "POST", "path": "/xrefs_to_batch", "desc": '{"targets": [...]} -- xrefs_to in one call'},
+    {"method": "GET", "path": "/reload", "desc": "Reload the plugin's code from disk without restarting IDA"},
+]
+
+
+class IDAHandler(BaseHTTPRequestHandler):
+    def log_message(self, fmt, *args):
+        pass
+
+    def send_json(self, data, code=200):
+        body = json.dumps(data, ensure_ascii=False).encode("utf-8")
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def read_body(self):
+        length = int(self.headers.get("Content-Length", 0))
+        return json.loads(self.rfile.read(length)) if length else {}
+
+    def parse_path(self):
+        parsed = urlparse(self.path)
+        qs = parse_qs(parsed.query)
+        return parsed.path, {k: v[0] for k, v in qs.items()}
+
+    def do_GET(self):
+        try:
+            if not _check_auth(self):
+                self.send_json({"error": "unauthorized", "code": "unauthorized"}, 401)
+                return
+
+            path, qs = self.parse_path()
+
+            if path == "/ping":
+                self.send_json(_ping())
+            elif path == "/help":
+                self.send_json({"instance": INSTANCE_ID, "port": PORT, "routes": HELP_ROUTES})
+            elif path == "/functions":
+                self.send_json(
+                    _list_functions(
+                        filter_str=qs.get("filter"),
+                        limit=int(qs.get("limit", 200)),
+                        offset=int(qs.get("offset", 0)),
+                    )
+                )
+            elif path == "/functions/search":
+                self.send_json(_search_functions(q=qs.get("q", ""), limit=int(qs.get("limit", 100))))
+            elif path == "/strings/search":
+                self.send_json(_search_strings(q=qs.get("q", ""), limit=int(qs.get("limit", 50))))
+            elif path == "/imports":
+                self.send_json(_get_imports(filter_str=qs.get("filter")))
+            elif path == "/segments":
+                self.send_json(_get_segments())
+            elif path == "/entry_points":
+                self.send_json(_get_entry_points())
+            elif path.startswith("/name/"):
+                self.send_json(_get_name_value(path[len("/name/") :]))
+            elif path == "/name_cache/rebuild":
+                self.send_json(_rebuild_name_cache())
+            elif path.startswith("/decompile/"):
+                self.send_json(_decompile(path[len("/decompile/") :]))
+            elif path.startswith("/disasm/"):
+                self.send_json(
+                    _get_disasm(
+                        path[len("/disasm/") :], count=int(qs.get("count", 30)), offset=int(qs.get("offset", 0))
+                    )
+                )
+            elif path.startswith("/xrefs_to/"):
+                self.send_json(_xrefs_to(path[len("/xrefs_to/") :]))
+            elif path.startswith("/xrefs_from/"):
+                self.send_json(_xrefs_from(path[len("/xrefs_from/") :]))
+            elif path.startswith("/prototype/"):
+                self.send_json(_get_prototype(path[len("/prototype/") :]))
+            elif path.startswith("/notes/"):
+                self.send_json(_get_notes(path[len("/notes/") :]))
+            elif path.startswith("/bytes/"):
+                self.send_json(_get_bytes(path[len("/bytes/") :], size=int(qs.get("size", 64))))
+            elif path == "/find":
+                self.send_json(_find_pattern(qs.get("pattern", ""), max_results=int(qs.get("limit", 50))))
+            elif path.startswith("/line_comment/"):
+                self.send_json(
+                    _get_line_comment(path[len("/line_comment/") :], repeatable=qs.get("repeatable") == "1")
+                )
+            elif path == "/top_functions":
+                self.send_json(_top_functions(by=qs.get("by", "xrefs"), limit=int(qs.get("limit", 20))))
+            elif path.startswith("/call_graph/"):
+                self.send_json(
+                    _call_graph(
+                        path[len("/call_graph/") :],
+                        depth=int(qs.get("depth", 2)),
+                        direction=qs.get("direction", "callees"),
+                    )
+                )
+            elif path == "/local_types":
+                self.send_json(_list_local_types(limit=int(qs.get("limit", 200))))
+            elif path == "/objc_classes":
+                self.send_json(_get_objc_classes(filter_str=qs.get("filter"), limit=int(qs.get("limit", 200))))
+            elif path == "/find_immediate":
+                self.send_json(_find_immediate(qs.get("value", "0"), limit=int(qs.get("limit", 100))))
+            elif path == "/find_vtables":
+                self.send_json(
+                    _find_vtables(
+                        segment_filter=qs.get("segment"),
+                        min_entries=int(qs.get("min_entries", 2)),
+                        limit=int(qs.get("limit", 50)),
+                    )
+                )
+            elif path == "/reload":
+                self.send_json(_reload_plugin_code())
+            else:
+                self.send_json({"error": "Unknown endpoint", "code": "not_found"}, 404)
+
+        except ApiError as e:
+            self.send_json({"error": str(e), "code": e.code}, 400)
+        except ValueError as e:
+            self.send_json({"error": str(e), "code": "bad_request"}, 400)
+        except Exception as e:
+            self.send_json({"error": str(e), "code": "internal_error"}, 500)
+
+    def do_POST(self):
+        try:
+            if not _check_auth(self):
+                self.send_json({"error": "unauthorized", "code": "unauthorized"}, 401)
+                return
+
+            body = self.read_body()
+            path, _ = self.parse_path()
+
+            if path == "/rename_function":
+                _require(body, ["target", "new_name"])
+                self.send_json(_rename_function(body["target"], body["new_name"]))
+            elif path == "/rename_local":
+                _require(body, ["function", "old_name", "new_name"])
+                self.send_json(_rename_local(body["function"], body["old_name"], body["new_name"]))
+            elif path == "/set_comment":
+                _require(body, ["target", "comment"])
+                self.send_json(_set_comment(body["target"], body["comment"]))
+            elif path == "/notes":
+                _require(body, ["target", "note"])
+                self.send_json(_set_notes(body["target"], body["note"]))
+            elif path == "/set_prototype":
+                _require(body, ["target", "prototype"])
+                self.send_json(_set_prototype(body["target"], body["prototype"]))
+            elif path == "/set_lvar_type":
+                _require(body, ["function", "lvar_name", "type"])
+                self.send_json(_set_lvar_type(body["function"], body["lvar_name"], body["type"]))
+            elif path == "/decompile_batch":
+                _require(body, ["targets"])
+                self.send_json(_decompile_many(body["targets"]))
+            elif path == "/patch_bytes":
+                _require(body, ["target", "hex_bytes"])
+                self.send_json(_patch_bytes(body["target"], body["hex_bytes"]))
+            elif path == "/make_data":
+                _require(body, ["target", "data_type"])
+                self.send_json(_make_data(body["target"], body["data_type"], body.get("struct_name")))
+            elif path == "/line_comment":
+                _require(body, ["target", "comment"])
+                self.send_json(_set_line_comment(body["target"], body["comment"], bool(body.get("repeatable", False))))
+            elif path == "/make_function":
+                _require(body, ["target"])
+                self.send_json(_make_function(body["target"]))
+            elif path == "/undefine":
+                _require(body, ["target"])
+                self.send_json(_undefine(body["target"]))
+            elif path == "/create_struct":
+                _require(body, ["name", "fields"])
+                self.send_json(_create_struct(body["name"], body["fields"]))
+            elif path == "/create_type":
+                _require(body, ["decl"])
+                self.send_json(_create_local_type(body["decl"]))
+            elif path == "/xrefs_to_batch":
+                _require(body, ["targets"])
+                self.send_json(_xrefs_to_many(body["targets"]))
+            else:
+                self.send_json({"error": "Unknown endpoint", "code": "not_found"}, 404)
+
+        except ApiError as e:
+            self.send_json({"error": str(e), "code": e.code}, 400)
+        except ValueError as e:
+            self.send_json({"error": str(e), "code": "bad_request"}, 400)
+        except Exception as e:
+            self.send_json({"error": str(e), "code": "internal_error"}, 500)
+
+
+class ThreadingHTTPServer(socketserver.ThreadingMixIn, HTTPServer):
+    daemon_threads = True
+
+
+# ─────────────────────────────────────────────
+#  IDA Plugin
+# ─────────────────────────────────────────────
+
+
+class IdaMcpPlugin(idaapi.plugin_t):
+    flags = idaapi.PLUGIN_KEEP
+    comment = "IDA MCP REST server (auto-port + self-registration)"
+    help = "Exposes IDA Pro via HTTP for Claude/any MCP client"
+    wanted_name = "IDA MCP"
+    wanted_hotkey = "Ctrl-Shift-M"
+
+    def init(self):
+        self.server = None
+        self.thread = None
+        self.heartbeat_stop = None
+        self.heartbeat_thread = None
+        self._start()
+        threading.Thread(target=self._build_cache_bg, daemon=True).start()
+        return idaapi.PLUGIN_KEEP
+
+    def _build_cache_bg(self):
+        time.sleep(3)  # give IDA time to finish loading the database
+        try:
+            run_in_main_thread(_build_name_cache)
+        except Exception as e:
+            print(f"[IDA MCP:{PORT}] Cache build error: {e}")
+
+    def run(self, arg):
+        if self.server:
+            self._stop()
+        else:
+            self._start()
+
+    def _start(self):
+        global PORT, INSTANCE_ID
+        try:
+            forced = os.environ.get("IDA_MCP_PORT")
+            PORT = int(forced) if forced else _find_free_port(BASE_PORT, BIND_HOST)
+            INSTANCE_ID = _make_instance_id()
+
+            self.server = ThreadingHTTPServer((BIND_HOST, PORT), IDAHandler)
+            self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+            self.thread.start()
+
+            self.heartbeat_stop = threading.Event()
+            self.heartbeat_thread = threading.Thread(
+                target=_heartbeat_loop, args=(self.heartbeat_stop,), daemon=True
+            )
+            self.heartbeat_thread.start()
+
+            print(
+                f"[IDA MCP] Started: id={INSTANCE_ID} http://{ADVERTISE_HOST}:{PORT} "
+                f"(bind {BIND_HOST}:{PORT}), registry={REGISTRY_URL or 'off'}"
+            )
+        except Exception as e:
+            print(f"[IDA MCP] Failed to start: {e}")
+
+    def _stop(self):
+        if self.heartbeat_stop:
+            self.heartbeat_stop.set()
+        _registry_call("/instances/unregister", {"id": INSTANCE_ID})
+        if self.server:
+            self.server.shutdown()
+            self.server = None
+            print(f"[IDA MCP:{PORT}] Stopped.")
+
+    def term(self):
+        self._stop()
+
+
+def PLUGIN_ENTRY():
+    return IdaMcpPlugin()