# Copyright (c) 2026 Vector 35 Inc
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
"""
Tools for Binary Ninja's Model Context Protocol (MCP) server.
A tool registered here is offered by every MCP server in the process, alongside the built-in ``bn_*``
tools. The simplest way to write one is the :py:func:`tool` decorator, which builds the tool's JSON
Schema from the function's signature and docstring::
from binaryninja import mcp
@mcp.tool(read_only=True)
def myplugin_function_count(call: mcp.ToolCall, min_size: int = 0) -> dict:
\"\"\"Count the functions in the active binary view.
:param min_size: Only count functions with at least this many bytes.
\"\"\"
functions = [f for f in call.binary_view.functions if f.total_bytes >= min_size]
return {"count": len(functions)}
"""
import ctypes
import enum
import inspect
import json
import re
import types
import typing
from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union
import binaryninja
from . import _binaryninjacore as core
from . import binaryview
from .enums import McpToolAnnotation, McpToolScope
from .log import log_error_for_exception
__all__ = [
"Address",
"ClampTo",
"IntegerExpression",
"Maximum",
"Minimum",
"NonEmpty",
"RelativeTo",
"Schema",
"Tool",
"ToolCall",
"ToolError",
"ToolResult",
"register_tool",
"tool",
]
[docs]
class Address(int):
"""
A parameter annotated ``Address`` accepts an address expression string, evaluated against the call's
binary view. The function receives an ``int``.
"""
__slots__ = ()
[docs]
class IntegerExpression(int):
"""
A parameter annotated ``IntegerExpression`` accepts an unsigned integer or an expression string,
evaluated against the call's binary view. The function receives an ``int``.
"""
__slots__ = ()
_TOOL_NAME = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
# The origin of a T | None annotation, which is not typing.Union on Python 3.10 and later.
_UnionType = getattr(types, "UnionType", Union)
# Keeps each registered tool's ctypes callbacks alive until the tool is freed.
_registered_tools: List["_RegisteredTool"] = []
[docs]
class Schema:
"""
Supplies the JSON Schema for a parameter the other annotations cannot describe. Use it as
``Annotated[dict, mcp.Schema({...})]``. The function receives the argument as decoded JSON.
"""
[docs]
def __init__(self, schema: dict):
self.schema = schema
[docs]
class RelativeTo:
"""
Evaluates an :py:class:`IntegerExpression` parameter's expression with an earlier :py:class:`Address`,
:py:class:`IntegerExpression` or ``int`` parameter's value as ``$here``. When that argument is absent,
its default is used, or 0 when it has none. Use it as
``Annotated[mcp.IntegerExpression, mcp.RelativeTo("address")]``, for example for a length measured
from an address, or on the items of a ``List`` of them.
"""
[docs]
def __init__(self, parameter: str):
self.parameter = parameter
[docs]
class Minimum:
"""Rejects an ``int`` parameter's values below ``value``. Use it as ``Annotated[int, mcp.Minimum(0)]``."""
[docs]
def __init__(self, value: int):
self.value = value
[docs]
class Maximum:
"""Rejects an ``int`` parameter's values above ``value``. Use it as ``Annotated[int, mcp.Maximum(100)]``."""
[docs]
def __init__(self, value: int):
self.value = value
[docs]
class ClampTo:
"""
Clamps an ``int`` parameter's value to the range ``minimum`` to ``maximum`` instead of rejecting values
outside it. Use it as ``Annotated[int, mcp.ClampTo(0, 1000)]``.
"""
[docs]
def __init__(self, minimum: int, maximum: int):
self.minimum = minimum
self.maximum = maximum
[docs]
class NonEmpty:
"""Rejects an empty ``str`` parameter. Use it as ``Annotated[str, mcp.NonEmpty()]``."""
class _ToolMetaclass(type):
def __iter__(cls):
binaryninja._init_plugins()
count = ctypes.c_ulonglong()
tools = core.BNGetMcpToolList(count)
try:
for i in range(count.value):
yield Tool(core.BNNewMcpToolReference(tools[i]))
finally:
core.BNFreeMcpToolList(tools, count.value)
class _RegisteredTool:
def __init__(self, handler: Callable[[ToolCall, dict], Any]):
self.handler = handler
self.callbacks = core.BNMcpToolCallbacks()
self.callbacks.context = 0
self.callbacks.invoke = self.callbacks.invoke.__class__(self._invoke)
self.callbacks.freeObject = self.callbacks.freeObject.__class__(self._free_object)
def _free_object(self, ctxt):
try:
_registered_tools.remove(self)
except Exception:
log_error_for_exception("Unhandled Python exception freeing an MCP tool")
def _invoke(self, ctxt, call_handle, arguments, result_handle):
try:
try:
decoded = json.loads(core.pyNativeStr(arguments))
result = ToolResult._from_value(self.handler(ToolCall(call_handle), decoded))
except ToolError as error:
result = ToolResult.from_error(error)
result._apply(result_handle)
except Exception as error:
log_error_for_exception("Unhandled Python exception in MCP tool")
try:
ToolResult.from_error(ToolError("internal_error", str(error)))._apply(result_handle)
except Exception:
log_error_for_exception("Unhandled Python exception applying an MCP tool result")
# Converts one argument, given the arguments converted before it.
_Converter = Callable[[ToolCall, Any, Dict[str, Any]], Any]
class _Parameter:
def __init__(
self, name: str, schema: dict, kind: str, convert: _Converter, required: bool, default: Any,
relative_to: Optional[str]
):
self.name = name
self.schema = schema
self.kind = kind
self.convert = convert
self.required = required
self.default = default
self.relative_to = relative_to
def _is_integer(value: Any) -> bool:
return isinstance(value, int) and not isinstance(value, bool)
def _relative_to(annotation) -> Optional[str]:
"""
Returns the parameter named by an ``Annotated[IntegerExpression, RelativeTo(...)]`` annotation, or by
the item annotation of a ``List``.
"""
origin = typing.get_origin(annotation)
if origin in (list, List):
arguments = typing.get_args(annotation)
return _relative_to(arguments[0]) if arguments else None
if origin is not typing.Annotated:
return None
base, *metadata = typing.get_args(annotation)
names = [item.parameter for item in metadata if isinstance(item, RelativeTo)]
if not names:
return None
if base is not IntegerExpression:
raise TypeError("RelativeTo only applies to parameters annotated mcp.IntegerExpression")
return names[0]
def _describe(annotation, name: str, relative_to: Optional[str] = None) -> Tuple[dict, str, _Converter]:
"""Returns the schema, the kind named in error messages, and the conversion for one annotation."""
def expect(kind: str, check: Callable[[Any], bool], convert: Callable[[Any], Any] = lambda value: value):
def run(call: ToolCall, value: Any, converted: Dict[str, Any]) -> Any:
if not check(value):
raise ToolError("invalid_params", f"Expected {kind} parameter '{name}'")
return convert(value)
return run
if annotation is Address:
def address(call: ToolCall, value: Any, converted: Dict[str, Any]) -> int:
try:
return call.parse_address(value)
except ValueError as error:
raise ToolError("invalid_params", f"Invalid address expression parameter '{name}': {error}")
return {"type": "string"}, "address expression", address
if annotation is IntegerExpression:
def integer(call: ToolCall, value: Any, converted: Dict[str, Any]) -> int:
here = converted.get(relative_to) if relative_to is not None else None
try:
# A negative int parameter used as $here wraps to 64 bits.
return call.parse_integer(value, (here or 0) & 0xFFFFFFFFFFFFFFFF)
except ValueError as error:
raise ToolError("invalid_params", f"Invalid unsigned integer parameter '{name}': {error}")
return {"type": ["integer", "string"], "minimum": 0}, "unsigned integer", integer
if annotation is str:
return {"type": "string"}, "string", expect("string", lambda value: isinstance(value, str))
if annotation is bool:
return {"type": "boolean"}, "boolean", expect("boolean", lambda value: isinstance(value, bool))
if annotation is int:
return {"type": "integer"}, "integer", expect("integer", _is_integer)
if annotation is float:
return (
{"type": "number"},
"number",
expect("number", lambda value: _is_integer(value) or isinstance(value, float), float),
)
if inspect.isclass(annotation) and issubclass(annotation, enum.Enum):
names = [member.name for member in annotation]
def member(call: ToolCall, value: Any, converted: Dict[str, Any]):
if not isinstance(value, str):
raise ToolError("invalid_params", f"Expected string parameter '{name}'")
if value not in names:
raise ToolError("invalid_params", f"Invalid enum value for parameter '{name}'")
return annotation[value]
return {"type": "string", "enum": names}, "string", member
origin = typing.get_origin(annotation)
arguments = typing.get_args(annotation)
if origin is Literal:
choices = list(arguments)
if not all(isinstance(choice, str) for choice in choices):
raise TypeError(f"Parameter '{name}': only string Literal choices are supported")
def choice(call: ToolCall, value: Any, converted: Dict[str, Any]) -> str:
if not isinstance(value, str):
raise ToolError("invalid_params", f"Expected string parameter '{name}'")
if value not in choices:
raise ToolError("invalid_params", f"Invalid enum value for parameter '{name}'")
return value
return {"type": "string", "enum": choices}, "string", choice
if origin in (list, List):
item_schema, item_kind, item_convert = _describe(arguments[0] if arguments else str, name, relative_to)
def items(call: ToolCall, value: Any, converted: Dict[str, Any]) -> list:
if not isinstance(value, list):
raise ToolError("invalid_params", f"Expected {item_kind} array parameter '{name}'")
return [item_convert(call, item, converted) for item in value]
return {"type": "array", "items": item_schema}, f"{item_kind} array", items
if origin is typing.Annotated:
base, *metadata = arguments
for item in metadata:
if isinstance(item, Schema):
return dict(item.schema), "JSON", lambda call, value, converted: value
limits = [item for item in metadata if isinstance(item, (Minimum, Maximum, ClampTo))]
if limits:
return _describe_limited_int(base, name, limits)
if any(item is NonEmpty or isinstance(item, NonEmpty) for item in metadata):
if base is not str:
raise TypeError(f"Parameter '{name}': NonEmpty only applies to str")
return (
{"type": "string", "minLength": 1},
"non-empty string",
expect("non-empty string", lambda value: isinstance(value, str) and value != ""),
)
return _describe(base, name, relative_to)
raise TypeError(f"Parameter '{name}': unsupported annotation {annotation!r}")
def _describe_limited_int(base, name: str, limits: list) -> Tuple[dict, str, _Converter]:
if base is not int:
raise TypeError(f"Parameter '{name}': Minimum, Maximum and ClampTo only apply to int")
minimum: Optional[int] = None
maximum: Optional[int] = None
clamp = False
for limit in limits:
if isinstance(limit, Minimum):
minimum = limit.value
elif isinstance(limit, Maximum):
maximum = limit.value
else:
minimum, maximum, clamp = limit.minimum, limit.maximum, True
schema: Dict[str, Any] = {"type": "integer"}
if minimum is not None:
schema["minimum"] = minimum
if maximum is not None:
schema["maximum"] = maximum
def convert(call: ToolCall, value: Any, converted: Dict[str, Any]) -> int:
if not _is_integer(value):
raise ToolError("invalid_params", f"Expected integer parameter '{name}'")
if minimum is not None and value < minimum:
if clamp:
return minimum
raise ToolError("invalid_params", f"Invalid integer parameter '{name}': Must be at least {minimum}")
if maximum is not None and value > maximum:
if clamp:
return maximum
raise ToolError("invalid_params", f"Invalid integer parameter '{name}': Must be at most {maximum}")
return value
return schema, "integer", convert
def _optional_inner(annotation):
"""Returns T for Optional[T] or T | None, or None when the annotation is not optional."""
origin = typing.get_origin(annotation)
if origin is Union or origin is _UnionType:
arguments = [argument for argument in typing.get_args(annotation) if argument is not type(None)]
if len(arguments) == 1 and len(typing.get_args(annotation)) == 2:
return arguments[0]
return None
def _parse_docstring(docstring: Optional[str]) -> Tuple[str, Dict[str, str]]:
"""Returns the leading paragraph and the ``:param name:`` descriptions."""
text = inspect.cleandoc(docstring or "")
# The description ends at the first blank line or ":field:" line. A field's continuation lines are
# indented and non-blank.
summary = re.split(r"\n\s*\n|\n(?=:)", text, maxsplit=1)[0]
description = " ".join(line.strip() for line in summary.splitlines())
params: Dict[str, str] = {}
for match in re.finditer(r"^:param\s+(\w+):\s*(.*(?:\n[ \t]+\S.*)*)", text, re.MULTILINE):
params[match.group(1)] = " ".join(part.strip() for part in match.group(2).splitlines())
return description, params
def _default_json(annotation, value: Any) -> Any:
"""Returns a default value in the form the parameter's schema describes."""
if annotation is Address:
return hex(value)
if isinstance(value, enum.Enum):
return value.name
origin = typing.get_origin(annotation)
arguments = typing.get_args(annotation)
if origin in (list, List):
return [_default_json(arguments[0] if arguments else str, item) for item in value]
if origin is typing.Annotated:
return _default_json(arguments[0], value)
return value
def _parameters(function: Callable, descriptions: Dict[str, str]) -> List[_Parameter]:
signature = inspect.signature(function)
hints = typing.get_type_hints(function, include_extras=True)
parameters = list(signature.parameters.values())
if not parameters:
raise TypeError(f"MCP tool '{function.__name__}' must take the ToolCall as its first parameter")
result: List[_Parameter] = []
for parameter in parameters[1:]:
name = parameter.name
if parameter.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
raise TypeError(f"MCP tool '{function.__name__}': *args and **kwargs are not supported")
if parameter.kind == inspect.Parameter.POSITIONAL_ONLY:
raise TypeError(f"MCP tool '{function.__name__}': parameter '{name}' cannot be positional-only")
if name not in hints:
raise TypeError(f"MCP tool '{function.__name__}': parameter '{name}' needs a type annotation")
if name not in descriptions:
raise TypeError(f"MCP tool '{function.__name__}': parameter '{name}' needs a ':param {name}:' description")
annotation = hints[name]
inner = _optional_inner(annotation)
target = inner if inner is not None else annotation
relative_to = _relative_to(target)
if relative_to is not None and not any(
earlier.name == relative_to and earlier.kind in ("address expression", "unsigned integer", "integer")
for earlier in result
):
raise TypeError(
f"MCP tool '{function.__name__}': parameter '{name}' is relative to '{relative_to}', which must be an "
"earlier integer parameter"
)
schema, kind, convert = _describe(target, name, relative_to)
schema = dict(schema)
schema["description"] = descriptions[name]
if has_default := parameter.default is not inspect.Parameter.empty:
if parameter.default is not None:
schema["default"] = _default_json(target, parameter.default)
required = inner is None and not has_default
default = parameter.default if has_default else None
result.append(_Parameter(name, schema, kind, convert, required, default, relative_to))
return result