Add: get_callers & get_callees tools (#58)

- `get_callers`: Takes one or more function names/addresses, returns JSON
{"results":[{identifier,function,callers,caller_sites}], "errors":[...]}
where each entry includes normalized function metadata, every caller,
and HLIL/IL snippets for each call site.
- `get_callees`: Accepts the same identifier inputs, returns
{"results":[{identifier,function,callees,call_sites}], "errors":[...]}
listing every outgoing callee plus per-site IL context, falling back to
raw addresses when no symbol exists.

Author: CX330Blake

README.md

@@ -189,6 +189,8 @@ The following table lists the available MCP functions for use:
 | `get_xrefs_to_union(union_name)`                                     | Get xrefs/usages related to a union (members, globals, code refs).                                           |
 | `get_stack_frame_vars(function_identifier)`                          | Get stack frame variable information for a function (names, offsets, sizes, types).                           |
 | `get_type_info(type_name)`                                           | Resolve a type and return declaration, kind, and members.                                                    |
+| `get_callers(identifiers)`                                           | List callers plus call sites for one or more function identifiers.                                           |
+| `get_callees(identifiers)`                                           | List callees plus call sites for one or more function identifiers.                                           |
 | `make_function_at(address, platform)`                                | Create a function at an address. `platform` optional; use `default` to pick the BinaryView/platform default. |
 | `list_platforms()`                                                   | List all available platform names.                                                                           |
 | `list_binaries()`                                                    | List managed/open binaries with ids and active flag.                                                         |
@@ -236,6 +238,8 @@ These are the list of HTTP endpoints that can be called:
 - `/getTypeInfo?name=<type>`: Resolve a type and return declaration and details.
 - `/getXrefsToUnion?name=<union>`: Union xrefs/usages (members, globals, refs).
 - `/getStackFrameVars?name=<function>|address=<addr>`: Get stack frame variable information for a function.
+- `/getCallers?identifiers=<name|addr>[,...]`: Return caller summaries (functions, call sites, HLIL/IL snippets) for one or more identifiers. Accepts `identifiers`, `identifier`, `names`, or `addresses` query params.
+- `/getCallees?identifiers=<name|addr>[,...]`: Return callee summaries with the same schema as `/getCallers`, detailing every outgoing call target per request identifier.
 - `/localTypes?offset=<n>&limit=<m>`: List local types.
 - `/strings?offset=<n>&limit=<m>`: Paginated strings.
 - `/strings/filter?offset=<n>&limit=<m>&filter=<substr>`: Filtered strings.

bridge/binja_mcp_bridge.py

@@ -805,6 +805,53 @@ def get_type_info(type_name: str) -> str:
     return _json.dumps(data, indent=2, ensure_ascii=False)
 
 
+def _normalize_identifier_input(value: str | list[str]) -> list[str]:
+    tokens: list[str] = []
+    if isinstance(value, str):
+        raw = value.replace(";", ",").split(",")
+        tokens.extend([tok.strip() for tok in raw if tok.strip()])
+    elif isinstance(value, (list, tuple, set)):
+        for item in value:
+            if item is None:
+                continue
+            tokens.extend(_normalize_identifier_input(str(item)))
+    return tokens
+
+
+@mcp.tool()
+def get_callers(identifiers: str) -> str:
+    """
+    List callers and caller sites for one or more function identifiers (name or address).
+    Provide comma-separated identifiers like "sub_401000,main".
+    """
+    items = _normalize_identifier_input(identifiers)
+    if not items:
+        return "Error: provide at least one identifier"
+    data = get_json("getCallers", {"identifiers": ",".join(items)}, timeout=None)
+    if not data:
+        return "Error: no response"
+    import json as _json
+
+    return _json.dumps(data, indent=2, ensure_ascii=False)
+
+
+@mcp.tool()
+def get_callees(identifiers: str) -> str:
+    """
+    List callees and call sites for one or more function identifiers (name or address).
+    Provide comma-separated identifiers like "sub_401000,main".
+    """
+    items = _normalize_identifier_input(identifiers)
+    if not items:
+        return "Error: provide at least one identifier"
+    data = get_json("getCallees", {"identifiers": ",".join(items)}, timeout=None)
+    if not data:
+        return "Error: no response"
+    import json as _json
+
+    return _json.dumps(data, indent=2, ensure_ascii=False)
+
+
 @mcp.tool()
 def set_function_prototype(name_or_address: str, prototype: str) -> str:
     """

plugin/api/endpoints.py

@@ -104,6 +104,14 @@ def get_function_info(self, identifier: str) -> dict[str, Any] | None:
             bn.log_error(f"Error getting function info: {e}")
             return None
 
+    def get_callers(self, identifiers: list[str]) -> dict[str, Any]:
+        """Proxy BinaryOperations.get_callers for endpoint reuse."""
+        return self.binary_ops.get_callers(identifiers)
+
+    def get_callees(self, identifiers: list[str]) -> dict[str, Any]:
+        """Proxy BinaryOperations.get_callees for endpoint reuse."""
+        return self.binary_ops.get_callees(identifiers)
+
     def get_imports(self, offset: int = 0, limit: int = 100) -> list[dict[str, Any]]:
         """Get list of imported functions"""
         if not self.binary_ops.current_view:

plugin/core/binary_operations.py

@@ -350,6 +350,204 @@ def get_function_by_name_or_address(self, identifier: str | int) -> bn.Function
         bn.log_error(f"Could not find function: {identifier}")
         return None
 
+    def _normalize_identifier_list(self, identifiers: Any) -> list[Any]:
+        """Normalize comma-delimited strings or iterables into a list of identifiers."""
+        if identifiers is None:
+            return []
+        if isinstance(identifiers, (list, tuple, set)):
+            raw_items = list(identifiers)
+        else:
+            raw_items = [identifiers]
+        normalized: list[Any] = []
+        for item in raw_items:
+            if item is None:
+                continue
+            if isinstance(item, str):
+                # Allow comma or semicolon separation for convenience
+                tokens = [tok.strip() for tok in item.replace(";", ",").split(",")]
+                normalized.extend([tok for tok in tokens if tok])
+            else:
+                normalized.append(item)
+        return normalized
+
+    def _format_function_reference(self, func: bn.Function | None) -> dict[str, Any] | None:
+        if not func:
+            return None
+        try:
+            return {
+                "name": getattr(func, "name", None),
+                "address": hex(int(func.start)) if hasattr(func, "start") else None,
+            }
+        except Exception:
+            return {
+                "name": getattr(func, "name", None),
+                "address": None,
+            }
+
+    def _collect_related_functions(
+        self, func: bn.Function, relation_attr: str
+    ) -> list[dict[str, Any]]:
+        related: list[dict[str, Any]] = []
+        seen: set[int] = set()
+        try:
+            rel_iter = getattr(func, relation_attr, None)
+        except Exception:
+            rel_iter = None
+        if rel_iter is None:
+            return related
+        try:
+            for rel_func in list(rel_iter):
+                if not rel_func:
+                    continue
+                addr = None
+                try:
+                    addr = int(rel_func.start)
+                except Exception:
+                    addr = None
+                if addr is not None and addr in seen:
+                    continue
+                if addr is not None:
+                    seen.add(addr)
+                ref = self._format_function_reference(rel_func)
+                if ref:
+                    related.append(ref)
+        except Exception:
+            pass
+        return related
+
+    def _summarize_call_sites(self, func: bn.Function, relation: str) -> list[dict[str, Any]]:
+        entries: list[dict[str, Any]] = []
+        attr = "caller_sites" if relation == "callers" else "call_sites"
+        try:
+            sites = getattr(func, attr, None)
+        except Exception:
+            sites = None
+        if not sites:
+            return entries
+
+        def _extract_function(site: Any, names: tuple[str, ...]) -> bn.Function | None:
+            for name in names:
+                try:
+                    value = getattr(site, name, None)
+                except Exception:
+                    value = None
+                if value:
+                    return value
+            return None
+
+        for site in list(sites):
+            try:
+                entry: dict[str, Any] = {}
+                addr = getattr(site, "address", None)
+                if isinstance(addr, int):
+                    entry["address"] = hex(addr)
+                elif isinstance(addr, str) and addr:
+                    entry["address"] = addr
+
+                if relation == "callers":
+                    caller_func = _extract_function(site, ("function", "source_function", "caller"))
+                    ref = self._format_function_reference(caller_func)
+                    if ref:
+                        entry["caller"] = ref
+                else:
+                    callee_func = _extract_function(
+                        site, ("callee", "dest_function", "target_function")
+                    )
+                    ref = self._format_function_reference(callee_func)
+                    if ref:
+                        entry["callee"] = ref
+                    else:
+                        # Fall back to raw destination address when available
+                        dest = None
+                        for attr_name in ("dest", "target", "constant"):
+                            try:
+                                dest = getattr(site, attr_name)
+                            except Exception:
+                                dest = None
+                            if dest is not None:
+                                break
+                        if isinstance(dest, int):
+                            entry["callee"] = {"name": None, "address": hex(dest)}
+
+                # Attach textual representation for quick context
+                summary_text = None
+                for attr_name in ("hlil", "il"):
+                    try:
+                        val = getattr(site, attr_name, None)
+                    except Exception:
+                        val = None
+                    if val is not None:
+                        summary_text = str(val)
+                        break
+                if summary_text is None:
+                    summary_text = str(site)
+                entry["il"] = summary_text
+
+                entries.append(entry)
+            except Exception:
+                continue
+        return entries
+
+    def get_callers(self, identifiers: Any) -> dict[str, Any]:
+        """Collect caller information for the given function identifiers."""
+        if not self._current_view:
+            raise RuntimeError("No binary loaded")
+
+        items = self._normalize_identifier_list(identifiers)
+        if not items:
+            raise ValueError("No function identifiers provided")
+
+        results: list[dict[str, Any]] = []
+        errors: list[str] = []
+        for ident in items:
+            try:
+                func = self.get_function_by_name_or_address(ident)
+            except Exception as exc:
+                func = None
+                errors.append(f"{ident}: {exc}")
+            if not func:
+                errors.append(f"Function not found: {ident}")
+                continue
+            entry = {
+                "identifier": str(ident),
+                "function": self._format_function_reference(func),
+                "callers": self._collect_related_functions(func, "callers"),
+                "caller_sites": self._summarize_call_sites(func, "callers"),
+            }
+            results.append(entry)
+
+        return {"results": results, "errors": errors}
+
+    def get_callees(self, identifiers: Any) -> dict[str, Any]:
+        """Collect callee information for the given function identifiers."""
+        if not self._current_view:
+            raise RuntimeError("No binary loaded")
+
+        items = self._normalize_identifier_list(identifiers)
+        if not items:
+            raise ValueError("No function identifiers provided")
+
+        results: list[dict[str, Any]] = []
+        errors: list[str] = []
+        for ident in items:
+            try:
+                func = self.get_function_by_name_or_address(ident)
+            except Exception as exc:
+                func = None
+                errors.append(f"{ident}: {exc}")
+            if not func:
+                errors.append(f"Function not found: {ident}")
+                continue
+            entry = {
+                "identifier": str(ident),
+                "function": self._format_function_reference(func),
+                "callees": self._collect_related_functions(func, "callees"),
+                "call_sites": self._summarize_call_sites(func, "callees"),
+            }
+            results.append(entry)
+
+        return {"results": results, "errors": errors}
+
     def get_function_names(self, offset: int = 0, limit: int = 100) -> list[dict[str, str]]:
         """Get list of function names with addresses"""
         if not self._current_view:

plugin/server/http_server.py

@@ -78,6 +78,28 @@ def _parse_query_params(self) -> dict[str, str]:
         parsed_path = urllib.parse.urlparse(self.path)
         return dict(urllib.parse.parse_qsl(parsed_path.query))
 
+    def _extract_identifiers(self, params: dict[str, str]) -> list[str]:
+        """Normalize identifier-bearing query params into a list."""
+        candidates: list[str] = []
+        for key in (
+            "identifiers",
+            "identifier",
+            "functions",
+            "function",
+            "names",
+            "name",
+            "addresses",
+            "address",
+        ):
+            value = params.get(key)
+            if value:
+                candidates.append(value)
+        identifiers: list[str] = []
+        for raw in candidates:
+            tokens = [tok.strip() for tok in raw.replace(";", ",").split(",")]
+            identifiers.extend([tok for tok in tokens if tok])
+        return identifiers
+
     def _parse_post_params(self) -> dict[str, Any]:
         """Parse POST request parameters from various formats.
 
@@ -745,6 +767,44 @@ def _printable(b: int) -> str:
                 matches = self.endpoints.search_functions(search_term, offset, limit)
                 self._send_json_response({"matches": matches})
 
+            elif path == "/getCallers":
+                identifiers = self._extract_identifiers(params)
+                if not identifiers:
+                    self._send_json_response(
+                        {
+                            "error": "Missing identifier parameter",
+                            "help": "Provide ?identifier=<name|address> or comma-separated ?identifiers=a,b",
+                        },
+                        400,
+                    )
+                    return
+                try:
+                    payload = self.binary_ops.get_callers(identifiers)
+                except Exception as e:
+                    bn.log_error(f"Error handling getCallers: {e}")
+                    self._send_json_response({"error": str(e)}, 500)
+                else:
+                    self._send_json_response(payload)
+
+            elif path == "/getCallees":
+                identifiers = self._extract_identifiers(params)
+                if not identifiers:
+                    self._send_json_response(
+                        {
+                            "error": "Missing identifier parameter",
+                            "help": "Provide ?identifier=<name|address> or comma-separated ?identifiers=a,b",
+                        },
+                        400,
+                    )
+                    return
+                try:
+                    payload = self.binary_ops.get_callees(identifiers)
+                except Exception as e:
+                    bn.log_error(f"Error handling getCallees: {e}")
+                    self._send_json_response({"error": str(e)}, 500)
+                else:
+                    self._send_json_response(payload)
+
             elif path == "/decompile":
                 function_name = params.get("name") or params.get("functionName")
                 if not function_name: