WintermuteMCP

Wintermute MCP Server — Stateful Translation Layer for LLM Agents.

This MCP server exposes the full Wintermute hardware-security and pentesting library to autonomous LLM agents via the Model Context Protocol.

Architecture

Because LLM agents cannot hold Python memory references, this server maintains an internal ObjectRegistry that maps human-readable string IDs to live Python objects. Every creation tool returns a string ID; every action tool accepts one. The LLM never sees raw memory addresses — only stable, meaningful identifiers like "op:acme-pentest" or "dev:iot-camera".

Registry Flow ~~~~~~~~~~~~~ 1. Agent calls create_operation(name="acme-pentest") 2. Server instantiates Operation("acme-pentest"), stores it under "op:acme-pentest" in the registry, and returns the ID string. 3. Agent calls add_device(operation_id="op:acme-pentest", hostname="gw01") 4. Server looks up "op:acme-pentest" → Operation object, calls op.addDevice("gw01"), stores the new Device as "dev:gw01", and returns that ID. 5. Agent can later call get_device_info(device_id="dev:gw01") to inspect the object, or addVulnerability_Device(device_id="dev:gw01", ...) to mutate it.

Running

wintermute-mcp                         # SSE on 127.0.0.1:31337
wintermute-mcp --host 0.0.0.0 --port 9000
wintermute-mcp --transport stdio       # for pipe-based MCP clients

ObjectRegistry

In-process object store keyed by agent-friendly string IDs.

Source code in wintermute/WintermuteMCP.py

class ObjectRegistry:
    """In-process object store keyed by agent-friendly string IDs."""

    def __init__(self) -> None:
        self._objects: dict[str, Any] = {}
        self._types: dict[str, str] = {}  # id → human type label
        self._counters: dict[str, int] = {}

    # -- helpers ----------------------------------------------------------

    @staticmethod
    def _sanitize(hint: str) -> str:
        return re.sub(r"[^a-zA-Z0-9_.-]", "_", hint)[:48]

    def _make_id(self, prefix: str, hint: str) -> str:
        base = f"{prefix}:{self._sanitize(hint)}"
        if base not in self._objects:
            return base
        self._counters.setdefault(base, 1)
        self._counters[base] += 1
        return f"{base}:{self._counters[base]}"

    # -- public API -------------------------------------------------------

    def store(self, obj: Any, type_label: str, hint: str, *, prefix: str = "") -> str:
        """Persist *obj* and return a unique string ID for the LLM."""
        pfx = prefix or type_label
        oid = self._make_id(pfx, hint)
        self._objects[oid] = obj
        self._types[oid] = type_label
        return oid

    def get(self, oid: str) -> Any | None:
        """Retrieve the Python object behind *oid*, or ``None``."""
        return self._objects.get(oid)

    def get_typed(self, oid: str, expected: type[Any]) -> Any | None:
        """Return the object only if it is an instance of *expected*."""
        obj = self._objects.get(oid)
        if obj is not None and isinstance(obj, expected):
            return obj
        return None

    def delete(self, oid: str) -> bool:
        if oid in self._objects:
            del self._objects[oid]
            del self._types[oid]
            return True
        return False

    def list_all(self) -> dict[str, str]:
        """Return ``{id: type_label}`` for every registered object."""
        return dict(self._types)

get(oid)

Retrieve the Python object behind oid, or None.

Source code in wintermute/WintermuteMCP.py

def get(self, oid: str) -> Any | None:
    """Retrieve the Python object behind *oid*, or ``None``."""
    return self._objects.get(oid)

get_typed(oid, expected)

Return the object only if it is an instance of expected.

Source code in wintermute/WintermuteMCP.py

def get_typed(self, oid: str, expected: type[Any]) -> Any | None:
    """Return the object only if it is an instance of *expected*."""
    obj = self._objects.get(oid)
    if obj is not None and isinstance(obj, expected):
        return obj
    return None

list_all()

Return {id: type_label} for every registered object.

Source code in wintermute/WintermuteMCP.py

def list_all(self) -> dict[str, str]:
    """Return ``{id: type_label}`` for every registered object."""
    return dict(self._types)

store(obj, type_label, hint, *, prefix='')

Persist obj and return a unique string ID for the LLM.

Source code in wintermute/WintermuteMCP.py

def store(self, obj: Any, type_label: str, hint: str, *, prefix: str = "") -> str:
    """Persist *obj* and return a unique string ID for the LLM."""
    pfx = prefix or type_label
    oid = self._make_id(pfx, hint)
    self._objects[oid] = obj
    self._types[oid] = type_label
    return oid

addVulnerability_AWSAccount(aws_account_id, title, description=None, threat=None, cvss=None, risk_json=None, verified=None) async

Add a vulnerability to an AWS Account.

Use this for cloud-level findings like overly permissive IAM policies, public S3 buckets, or missing CloudTrail logging.

You MUST provide a valid aws_account_id obtained from add_aws_account().

Parameters:

Name	Type	Description	Default
aws_account_id	str	Registry ID of the target AWSAccount.	required
title	str	Short vulnerability title.	required
description	str | None	Detailed finding description.	None
threat	str | None	Threat level label.	None
cvss	int | None	CVSS score (integer 0-10).	None
risk_json	str | None	JSON object with likelihood, impact, severity.	None
verified	bool | None	Whether the vulnerability has been confirmed.	None

Returns:

Type	Description
str	Confirmation with the vulnerability ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def addVulnerability_AWSAccount(
    aws_account_id: str,
    title: str,
    description: str | None = None,
    threat: str | None = None,
    cvss: int | None = None,
    risk_json: str | None = None,
    verified: bool | None = None,
) -> str:
    """Add a vulnerability to an AWS Account.

    Use this for cloud-level findings like overly permissive IAM policies,
    public S3 buckets, or missing CloudTrail logging.

    You MUST provide a valid ``aws_account_id`` obtained from add_aws_account().

    Args:
        aws_account_id: Registry ID of the target AWSAccount.
        title: Short vulnerability title.
        description: Detailed finding description.
        threat: Threat level label.
        cvss: CVSS score (integer 0-10).
        risk_json: JSON object with ``likelihood``, ``impact``, ``severity``.
        verified: Whether the vulnerability has been confirmed.

    Returns:
        Confirmation with the vulnerability ID.
    """
    acc: AWSAccount = _require(aws_account_id, AWSAccount, "AWSAccount")
    return _add_vuln_to(acc, title, description, threat, cvss, risk_json, verified)

addVulnerability_Device(device_id, title, description=None, threat=None, cvss=None, risk_json=None, verified=None) async

Add a vulnerability directly to a Device (host-level finding).

Use this for vulnerabilities that affect the device itself rather than a specific service — for example, weak SSH host keys, missing patches, or exposed debug interfaces.

You MUST provide a valid device_id obtained from add_device().

Parameters:

Name	Type	Description	Default
device_id	str	Registry ID of the target Device.	required
title	str	Short vulnerability title.	required
description	str | None	Detailed finding description.	None
threat	str | None	Threat level label (e.g. "critical", "high").	None
cvss	int | None	CVSS score (integer 0-10).	None
risk_json	str | None	JSON object with likelihood, impact, severity.	None
verified	bool | None	Whether the vulnerability has been confirmed.	None

Returns:

Type	Description
str	Confirmation with the vulnerability ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def addVulnerability_Device(
    device_id: str,
    title: str,
    description: str | None = None,
    threat: str | None = None,
    cvss: int | None = None,
    risk_json: str | None = None,
    verified: bool | None = None,
) -> str:
    """Add a vulnerability directly to a Device (host-level finding).

    Use this for vulnerabilities that affect the device itself rather than a
    specific service — for example, weak SSH host keys, missing patches, or
    exposed debug interfaces.

    You MUST provide a valid ``device_id`` obtained from add_device().

    Args:
        device_id: Registry ID of the target Device.
        title: Short vulnerability title.
        description: Detailed finding description.
        threat: Threat level label (e.g. ``"critical"``, ``"high"``).
        cvss: CVSS score (integer 0-10).
        risk_json: JSON object with ``likelihood``, ``impact``, ``severity``.
        verified: Whether the vulnerability has been confirmed.

    Returns:
        Confirmation with the vulnerability ID.
    """
    from wintermute.core import Device

    dev: Device = _require(device_id, Device, "Device")
    return _add_vuln_to(dev, title, description, threat, cvss, risk_json, verified)

addVulnerability_Peripheral(peripheral_id, title, description=None, threat=None, cvss=None, risk_json=None, verified=None) async

Add a vulnerability to a hardware Peripheral (UART, JTAG, etc.).

Use this for hardware-level findings like unauthenticated UART root shell, exposed JTAG debug port, or SWD readout not disabled.

You MUST provide a valid peripheral_id obtained from add_peripheral_to_device().

Parameters:

Name	Type	Description	Default
peripheral_id	str	Registry ID of the target Peripheral.	required
title	str	Short vulnerability title.	required
description	str | None	Detailed finding description.	None
threat	str | None	Threat level label.	None
cvss	int | None	CVSS score (integer 0-10).	None
risk_json	str | None	JSON object with likelihood, impact, severity.	None
verified	bool | None	Whether the vulnerability has been confirmed.	None

Returns:

Type	Description
str	Confirmation with the vulnerability ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def addVulnerability_Peripheral(
    peripheral_id: str,
    title: str,
    description: str | None = None,
    threat: str | None = None,
    cvss: int | None = None,
    risk_json: str | None = None,
    verified: bool | None = None,
) -> str:
    """Add a vulnerability to a hardware Peripheral (UART, JTAG, etc.).

    Use this for hardware-level findings like unauthenticated UART root shell,
    exposed JTAG debug port, or SWD readout not disabled.

    You MUST provide a valid ``peripheral_id`` obtained from
    add_peripheral_to_device().

    Args:
        peripheral_id: Registry ID of the target Peripheral.
        title: Short vulnerability title.
        description: Detailed finding description.
        threat: Threat level label.
        cvss: CVSS score (integer 0-10).
        risk_json: JSON object with ``likelihood``, ``impact``, ``severity``.
        verified: Whether the vulnerability has been confirmed.

    Returns:
        Confirmation with the vulnerability ID.
    """
    from wintermute.basemodels import Peripheral

    periph: Peripheral = _require(peripheral_id, Peripheral, "Peripheral")
    return _add_vuln_to(periph, title, description, threat, cvss, risk_json, verified)

addVulnerability_Service(service_id, title, description=None, threat=None, cvss=None, risk_json=None, verified=None) async

Add a vulnerability to a network Service.

Use this for service-level findings like SQL injection, XSS, missing HSTS, or TLS misconfigurations.

You MUST provide a valid service_id obtained from add_service_to_device().

Parameters:

Name	Type	Description	Default
service_id	str	Registry ID of the target Service.	required
title	str	Short vulnerability title.	required
description	str | None	Detailed finding description.	None
threat	str | None	Threat level label.	None
cvss	int | None	CVSS score (integer 0-10).	None
risk_json	str | None	JSON object with likelihood, impact, severity.	None
verified	bool | None	Whether the vulnerability has been confirmed.	None

Returns:

Type	Description
str	Confirmation with the vulnerability ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def addVulnerability_Service(
    service_id: str,
    title: str,
    description: str | None = None,
    threat: str | None = None,
    cvss: int | None = None,
    risk_json: str | None = None,
    verified: bool | None = None,
) -> str:
    """Add a vulnerability to a network Service.

    Use this for service-level findings like SQL injection, XSS, missing HSTS,
    or TLS misconfigurations.

    You MUST provide a valid ``service_id`` obtained from add_service_to_device().

    Args:
        service_id: Registry ID of the target Service.
        title: Short vulnerability title.
        description: Detailed finding description.
        threat: Threat level label.
        cvss: CVSS score (integer 0-10).
        risk_json: JSON object with ``likelihood``, ``impact``, ``severity``.
        verified: Whether the vulnerability has been confirmed.

    Returns:
        Confirmation with the vulnerability ID.
    """
    from wintermute.core import Service

    svc: Service = _require(service_id, Service, "Service")
    return _add_vuln_to(svc, title, description, threat, cvss, risk_json, verified)

addVulnerability_User(user_id, title, description=None, threat=None, cvss=None, risk_json=None, verified=None) async

Add a vulnerability to a User account.

Use this for user-level findings like weak passwords, credential reuse, excessive permissions, or successful phishing.

You MUST provide a valid user_id obtained from add_user().

Parameters:

Name	Type	Description	Default
user_id	str	Registry ID of the target User.	required
title	str	Short vulnerability title.	required
description	str | None	Detailed finding description.	None
threat	str | None	Threat level label.	None
cvss	int | None	CVSS score (integer 0-10).	None
risk_json	str | None	JSON object with likelihood, impact, severity.	None
verified	bool | None	Whether the vulnerability has been confirmed.	None

Returns:

Type	Description
str	Confirmation with the vulnerability ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def addVulnerability_User(
    user_id: str,
    title: str,
    description: str | None = None,
    threat: str | None = None,
    cvss: int | None = None,
    risk_json: str | None = None,
    verified: bool | None = None,
) -> str:
    """Add a vulnerability to a User account.

    Use this for user-level findings like weak passwords, credential reuse,
    excessive permissions, or successful phishing.

    You MUST provide a valid ``user_id`` obtained from add_user().

    Args:
        user_id: Registry ID of the target User.
        title: Short vulnerability title.
        description: Detailed finding description.
        threat: Threat level label.
        cvss: CVSS score (integer 0-10).
        risk_json: JSON object with ``likelihood``, ``impact``, ``severity``.
        verified: Whether the vulnerability has been confirmed.

    Returns:
        Confirmation with the vulnerability ID.
    """
    from wintermute.core import User

    user: User = _require(user_id, User, "User")
    return _add_vuln_to(user, title, description, threat, cvss, risk_json, verified)

add_analyst(operation_id, name, userid, email=None) async

Add a security analyst to an Operation.

Analysts are the team members performing the engagement.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
name	str	Full name of the analyst.	required
userid	str	Short username / ID.	required
email	str | None	Contact email.	None

Returns:

Type	Description
str	Confirmation message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_analyst(
    operation_id: str,
    name: str,
    userid: str,
    email: str | None = None,
) -> str:
    """Add a security analyst to an Operation.

    Analysts are the team members performing the engagement.

    Args:
        operation_id: Registry ID of the parent Operation.
        name: Full name of the analyst.
        userid: Short username / ID.
        email: Contact email.

    Returns:
        Confirmation message.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    op.addAnalyst(name, userid, email or "")
    return f"Analyst '{userid}' added to {operation_id}."

add_aws_account(operation_id, name, account_id=None, description=None, default_region=None, partition=None) async

Add an AWS account with IAM support to an Operation.

This creates an AWSAccount object which supports IAM users, IAM roles, AWS services, and per-account vulnerabilities.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
name	str	Account display name (e.g. "acme-prod").	required
account_id	str | None	AWS 12-digit account ID (e.g. "123456789012").	None
description	str | None	Free-text description.	None
default_region	str | None	Default AWS region (e.g. "us-east-1").	None
partition	str | None	AWS partition (default "aws").	None

Returns:

Type	Description
str	The AWS account's registry string ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_aws_account(
    operation_id: str,
    name: str,
    account_id: str | None = None,
    description: str | None = None,
    default_region: str | None = None,
    partition: str | None = None,
) -> str:
    """Add an AWS account with IAM support to an Operation.

    This creates an AWSAccount object which supports IAM users, IAM roles,
    AWS services, and per-account vulnerabilities.

    Args:
        operation_id: Registry ID of the parent Operation.
        name: Account display name (e.g. ``"acme-prod"``).
        account_id: AWS 12-digit account ID (e.g. ``"123456789012"``).
        description: Free-text description.
        default_region: Default AWS region (e.g. ``"us-east-1"``).
        partition: AWS partition (default ``"aws"``).

    Returns:
        The AWS account's registry string ID.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    kwargs: dict[str, Any] = {"name": name, "description": description or ""}
    if account_id:
        kwargs["account_id"] = account_id
    if default_region:
        kwargs["default_region"] = default_region
    if partition and partition != "aws":
        kwargs["partition"] = partition
    op.addAWSAccount(**kwargs)
    aws = op.awsaccounts[-1]
    aws_id = registry.store(aws, "AWSAccount", name, prefix="aws")
    return f"AWS account added. ID: {aws_id}"

add_cloud_account(operation_id, name, cloud_type=None, description=None, account_id=None) async

Add a generic cloud account to an Operation.

For AWS-specific accounts with IAM users/roles, prefer add_aws_account().

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
name	str	Account display name.	required
cloud_type	str | None	Provider type (e.g. "AWS", "GCP", "Azure").	None
description	str | None	Free-text description.	None
account_id	str | None	Cloud provider account ID.	None

Returns:

Type	Description
str	The cloud account's registry string ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_cloud_account(
    operation_id: str,
    name: str,
    cloud_type: str | None = None,
    description: str | None = None,
    account_id: str | None = None,
) -> str:
    """Add a generic cloud account to an Operation.

    For AWS-specific accounts with IAM users/roles, prefer add_aws_account().

    Args:
        operation_id: Registry ID of the parent Operation.
        name: Account display name.
        cloud_type: Provider type (e.g. ``"AWS"``, ``"GCP"``, ``"Azure"``).
        description: Free-text description.
        account_id: Cloud provider account ID.

    Returns:
        The cloud account's registry string ID.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    kwargs: dict[str, Any] = {
        "name": name,
        "cloud_type": cloud_type or "AWS",
        "description": description or "",
    }
    if account_id:
        kwargs["account_id"] = account_id
    op.addCloudAccount(**kwargs)
    acc = op.cloud_accounts[-1]
    acc_id = registry.store(acc, type(acc).__name__, name, prefix="cloud")
    return f"Cloud account added. ID: {acc_id}"

add_device(operation_id, hostname, ipaddr=None, macaddr=None, operating_system=None, fqdn=None) async

Add a network device or embedded target to an Operation.

You MUST provide a valid operation_id obtained from create_operation(). The device is stored in the Operation's device list AND registered separately in the session registry so you can reference it by its own ID.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
hostname	str	Unique hostname for the device (e.g. "iot-camera-v2").	required
ipaddr	str | None	IP address (IPv4 or IPv6 string). Defaults to "0.0.0.0".	None
macaddr	str | None	MAC address (defaults to "00:00:00:00:00:00").	None
operating_system	str | None	OS description (e.g. "Linux 5.10").	None
fqdn	str | None	Fully-qualified domain name.	None

Returns:

Type	Description
str	The device's registry string ID (e.g. "dev:iot-camera-v2").

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_device(
    operation_id: str,
    hostname: str,
    ipaddr: str | None = None,
    macaddr: str | None = None,
    operating_system: str | None = None,
    fqdn: str | None = None,
) -> str:
    """Add a network device or embedded target to an Operation.

    You MUST provide a valid ``operation_id`` obtained from create_operation().
    The device is stored in the Operation's device list AND registered
    separately in the session registry so you can reference it by its own ID.

    Args:
        operation_id: Registry ID of the parent Operation.
        hostname: Unique hostname for the device (e.g. ``"iot-camera-v2"``).
        ipaddr: IP address (IPv4 or IPv6 string). Defaults to ``"0.0.0.0"``.
        macaddr: MAC address (defaults to ``"00:00:00:00:00:00"``).
        operating_system: OS description (e.g. ``"Linux 5.10"``).
        fqdn: Fully-qualified domain name.

    Returns:
        The device's registry string ID (e.g. ``"dev:iot-camera-v2"``).
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    op.addDevice(
        hostname,
        ipaddr or "0.0.0.0",
        macaddr or "00:00:00:00:00:00",
        operating_system or "",
        fqdn or "",
    )
    dev = op.getDeviceByHostname(hostname)
    if dev is None:
        return f"addDevice returned success but device '{hostname}' not found."
    dev_id = registry.store(dev, "device", hostname, prefix="dev")
    return f"Device added. ID: {dev_id}"

add_iam_role_to_aws(aws_account_id, role_name, arn=None, administrator=None, policies_json=None) async

Add an IAM role to a registered AWS Account.

Parameters:

Name	Type	Description	Default
aws_account_id	str	Registry ID of the target AWSAccount.	required
role_name	str	IAM role name.	required
arn	str | None	IAM role ARN (optional).	None
administrator	bool | None	Whether this role has admin privileges.	None
policies_json	str | None	JSON array of attached policy ARNs.	None

Returns:

Type	Description
str	Confirmation message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_iam_role_to_aws(
    aws_account_id: str,
    role_name: str,
    arn: str | None = None,
    administrator: bool | None = None,
    policies_json: str | None = None,
) -> str:
    """Add an IAM role to a registered AWS Account.

    Args:
        aws_account_id: Registry ID of the target AWSAccount.
        role_name: IAM role name.
        arn: IAM role ARN (optional).
        administrator: Whether this role has admin privileges.
        policies_json: JSON array of attached policy ARNs.

    Returns:
        Confirmation message.
    """
    acc: AWSAccount = _require(aws_account_id, AWSAccount, "AWSAccount")
    try:
        policies: list[str] = json.loads(policies_json or "[]")
    except json.JSONDecodeError:
        policies = []
    acc.addIAMRole(
        role_name=role_name,
        arn=arn or None,
        administrator=administrator if administrator is not None else False,
        attached_policies=policies,
    )
    return f"IAM role '{role_name}' added to {aws_account_id}."

add_iam_user_to_aws(aws_account_id, username, arn=None, administrator=None, policies_json=None) async

Add an IAM user to a registered AWS Account.

Parameters:

Name	Type	Description	Default
aws_account_id	str	Registry ID of the target AWSAccount.	required
username	str	IAM username.	required
arn	str | None	IAM user ARN (optional).	None
administrator	bool | None	Whether this user has admin privileges.	None
policies_json	str | None	JSON array of attached policy ARNs.	None

Returns:

Type	Description
str	Confirmation message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_iam_user_to_aws(
    aws_account_id: str,
    username: str,
    arn: str | None = None,
    administrator: bool | None = None,
    policies_json: str | None = None,
) -> str:
    """Add an IAM user to a registered AWS Account.

    Args:
        aws_account_id: Registry ID of the target AWSAccount.
        username: IAM username.
        arn: IAM user ARN (optional).
        administrator: Whether this user has admin privileges.
        policies_json: JSON array of attached policy ARNs.

    Returns:
        Confirmation message.
    """
    acc: AWSAccount = _require(aws_account_id, AWSAccount, "AWSAccount")
    try:
        policies: list[str] = json.loads(policies_json or "[]")
    except json.JSONDecodeError:
        policies = []
    acc.addIAMUser(
        username=username,
        arn=arn or None,
        administrator=administrator if administrator is not None else False,
        attached_policies=policies,
    )
    return f"IAM user '{username}' added to {aws_account_id}."

add_note_to_test_run(test_run_id, note) async

Append a free-text note to an existing TestCaseRun.

The notes field on :class:TestCaseRun is a single string; this tool appends note to it on a fresh line so each note remains legible in the eventual report.

Parameters:

Name	Type	Description	Default
test_run_id	str	Registry ID of the target TestCaseRun (as returned by :func:generate_test_runs or :func:list_active_objects).	required
note	str	Text to append. Multi-line notes are honored verbatim.	required

Returns:

Type	Description
str	JSON object with run_id and the post-append notes_length,
str	or {"error": "..."} on registry miss.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_note_to_test_run(test_run_id: str, note: str) -> str:
    """Append a free-text note to an existing TestCaseRun.

    The ``notes`` field on :class:`TestCaseRun` is a single string; this
    tool appends ``note`` to it on a fresh line so each note remains
    legible in the eventual report.

    Args:
        test_run_id: Registry ID of the target TestCaseRun (as returned by
            :func:`generate_test_runs` or :func:`list_active_objects`).
        note: Text to append. Multi-line notes are honored verbatim.

    Returns:
        JSON object with ``run_id`` and the post-append ``notes_length``,
        or ``{"error": "..."}`` on registry miss.
    """
    try:
        run: TestCaseRun = _require(test_run_id, TestCaseRun, "TestCaseRun")
    except (ValueError, TypeError) as exc:
        return json.dumps({"error": str(exc)}, indent=2)
    run.notes = f"{run.notes}\n{note}" if run.notes else note
    return json.dumps(
        {"run_id": run.run_id, "notes_length": len(run.notes)},
        indent=2,
    )

add_peripheral_to_device(device_id, peripheral_type, name=None, device_path=None, pins_json=None, baudrate=None) async

Attach a hardware debug peripheral (UART, JTAG, etc.) to a Device.

Use this after identifying physical debug interfaces on an embedded target.

Parameters:

Name	Type	Description	Default
device_id	str	Registry ID of the target Device.	required
peripheral_type	str	One of "UART" or "JTAG". Other types can be added via the Wintermute library directly.	required
name	str | None	Human-readable label (e.g. "debug-uart").	None
device_path	str | None	OS device path (e.g. "/dev/ttyUSB0").	None
pins_json	str | None	JSON object mapping pin names to board locations (e.g. '{"tx":"J3-1","rx":"J3-2","gnd":"J3-3"}').	None
baudrate	int | None	Baud rate for UART peripherals (ignored for JTAG).	None

Returns:

Type	Description
str	The peripheral's registry string ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_peripheral_to_device(
    device_id: str,
    peripheral_type: str,
    name: str | None = None,
    device_path: str | None = None,
    pins_json: str | None = None,
    baudrate: int | None = None,
) -> str:
    """Attach a hardware debug peripheral (UART, JTAG, etc.) to a Device.

    Use this after identifying physical debug interfaces on an embedded target.

    Args:
        device_id: Registry ID of the target Device.
        peripheral_type: One of ``"UART"`` or ``"JTAG"``.  Other types can be
                         added via the Wintermute library directly.
        name: Human-readable label (e.g. ``"debug-uart"``).
        device_path: OS device path (e.g. ``"/dev/ttyUSB0"``).
        pins_json: JSON object mapping pin names to board locations
                   (e.g. ``'{"tx":"J3-1","rx":"J3-2","gnd":"J3-3"}'``).
        baudrate: Baud rate for UART peripherals (ignored for JTAG).

    Returns:
        The peripheral's registry string ID.
    """
    from wintermute.core import Device

    dev: Device = _require(device_id, Device, "Device")
    try:
        pins: dict[str, Any] = json.loads(pins_json or "{}")
    except json.JSONDecodeError:
        pins = {}

    ptype = peripheral_type.upper()
    label = name or f"{ptype.lower()}-{len(dev.peripherals)}"
    dpath = device_path or ""
    periph: UART | JTAG
    if ptype == "UART":
        periph = UART(
            device_path=dpath, name=label, pins=pins, baudrate=baudrate or 115200
        )
    elif ptype == "JTAG":
        periph = JTAG(device_path=dpath, name=label, pins=pins)
    else:
        return (
            f"Unsupported peripheral_type: '{peripheral_type}'. Use 'UART' or 'JTAG'."
        )

    dev.peripherals.append(periph)
    pid = registry.store(periph, "peripheral", label, prefix="periph")
    return f"Peripheral added. ID: {pid}"

add_reproduction_step_to_vulnerability(target_id, vuln_title, step_title, step_description=None, tool=None, action=None, confidence=None, arguments_json=None) async

Attach a reproduction step to a specific vulnerability on any target.

The target_id can be a Device, Service, AWSAccount, User, or Peripheral — any registry object that holds a vulnerabilities list. The vulnerability is looked up by vuln_title.

Parameters:

Name	Type	Description	Default
target_id	str	Registry ID of the object holding the vulnerability.	required
vuln_title	str	Exact title of the vulnerability to attach the step to.	required
step_title	str	Short title for this reproduction step.	required
step_description	str | None	Detailed description of what this step does.	None
tool	str | None	Tool or binary used (e.g. "nmap", "openocd").	None
action	str | None	Action performed by the tool (e.g. "scan", "read").	None
confidence	int | None	Confidence score 0-100.	None
arguments_json	str | None	JSON array of command-line arguments.	None

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_reproduction_step_to_vulnerability(
    target_id: str,
    vuln_title: str,
    step_title: str,
    step_description: str | None = None,
    tool: str | None = None,
    action: str | None = None,
    confidence: int | None = None,
    arguments_json: str | None = None,
) -> str:
    """Attach a reproduction step to a specific vulnerability on any target.

    The ``target_id`` can be a Device, Service, AWSAccount, User, or
    Peripheral — any registry object that holds a ``vulnerabilities`` list.
    The vulnerability is looked up by ``vuln_title``.

    Args:
        target_id: Registry ID of the object holding the vulnerability.
        vuln_title: Exact title of the vulnerability to attach the step to.
        step_title: Short title for this reproduction step.
        step_description: Detailed description of what this step does.
        tool: Tool or binary used (e.g. ``"nmap"``, ``"openocd"``).
        action: Action performed by the tool (e.g. ``"scan"``, ``"read"``).
        confidence: Confidence score 0-100.
        arguments_json: JSON array of command-line arguments.

    Returns:
        Confirmation or error message.
    """
    obj = registry.get(target_id)
    if obj is None:
        return f"ID '{target_id}' not found."
    if not hasattr(obj, "vulnerabilities"):
        return f"Object '{target_id}' does not support vulnerabilities."
    try:
        args_list: list[str] = json.loads(arguments_json or "[]")
    except json.JSONDecodeError:
        args_list = []
    step = ReproductionStep(
        title=step_title,
        description=step_description or "",
        tool=tool or None,
        action=action or None,
        confidence=confidence if confidence is not None else 0,
        arguments=args_list,
    )
    result = add_reproduction_step(obj, title=vuln_title, step=step)
    return f"Step '{step_title}' added to vulnerability '{vuln_title}': {result}"

add_service_to_aws(aws_account_id, name, arn=None, service_type=None, config_json=None) async

Add an AWS service resource to a registered AWS Account.

Parameters:

Name	Type	Description	Default
aws_account_id	str	Registry ID of the target AWSAccount.	required
name	str	Service display name.	required
arn	str | None	Full ARN of the service resource (provide when known).	None
service_type	str | None	AWS service type — one of ec2, s3, rds, lambda, iam, vpc, dynamodb, sqs, sns, elb, eks, ssm, ecr, other.	None
config_json	str | None	JSON object with service-specific configuration.	None

Returns:

Type	Description
str	Confirmation message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_service_to_aws(
    aws_account_id: str,
    name: str,
    arn: str | None = None,
    service_type: str | None = None,
    config_json: str | None = None,
) -> str:
    """Add an AWS service resource to a registered AWS Account.

    Args:
        aws_account_id: Registry ID of the target AWSAccount.
        name: Service display name.
        arn: Full ARN of the service resource (provide when known).
        service_type: AWS service type — one of ``ec2``, ``s3``, ``rds``,
                      ``lambda``, ``iam``, ``vpc``, ``dynamodb``, ``sqs``,
                      ``sns``, ``elb``, ``eks``, ``ssm``, ``ecr``, ``other``.
        config_json: JSON object with service-specific configuration.

    Returns:
        Confirmation message.
    """
    acc: AWSAccount = _require(aws_account_id, AWSAccount, "AWSAccount")
    try:
        stype = AWSServiceType((service_type or "other").lower())
    except ValueError:
        stype = AWSServiceType.OTHER
    try:
        config: dict[str, Any] = json.loads(config_json or "{}")
    except json.JSONDecodeError:
        config = {}
    acc.addService(name=name, arn=arn or "", service_type=stype, config=config)
    return f"AWS service '{name}' ({stype.value}) added to {aws_account_id}."

add_service_to_device(device_id, port_number, app=None, protocol=None, banner=None, transport_layer=None) async

Add a network service (open port) to a Device.

You MUST provide a valid device_id obtained from add_device().

Parameters:

Name	Type	Description	Default
device_id	str	Registry ID of the target Device.	required
port_number	int	TCP/UDP port number.	required
app	str | None	Application name (e.g. "nginx", "openssh").	None
protocol	str | None	Network protocol (default "ipv4").	None
banner	str | None	Service banner string captured during scanning.	None
transport_layer	str | None	Transport description (e.g. "HTTPS", "SSH").	None

Returns:

Type	Description
str	The service's registry string ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_service_to_device(
    device_id: str,
    port_number: int,
    app: str | None = None,
    protocol: str | None = None,
    banner: str | None = None,
    transport_layer: str | None = None,
) -> str:
    """Add a network service (open port) to a Device.

    You MUST provide a valid ``device_id`` obtained from add_device().

    Args:
        device_id: Registry ID of the target Device.
        port_number: TCP/UDP port number.
        app: Application name (e.g. ``"nginx"``, ``"openssh"``).
        protocol: Network protocol (default ``"ipv4"``).
        banner: Service banner string captured during scanning.
        transport_layer: Transport description (e.g. ``"HTTPS"``, ``"SSH"``).

    Returns:
        The service's registry string ID.
    """
    from wintermute.core import Device

    dev: Device = _require(device_id, Device, "Device")
    dev.addService(
        protocol=protocol or "ipv4",
        app=app or "",
        portNumber=port_number,
        banner=banner or "",
        transport_layer=transport_layer or "",
    )
    svc = dev.services[-1]
    svc_id = registry.store(
        svc, "service", f"{dev.hostname}:{port_number}", prefix="svc"
    )
    return f"Service added. ID: {svc_id}"

add_test_plan(operation_id, code, name, description=None, test_cases_json=None) async

Add a test plan to an Operation.

A test plan groups related test cases. Each test case defines a specific check to execute against a target (device, peripheral, cloud resource).

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
code	str	Unique plan code (e.g. "TP-UART-001").	required
name	str	Human-readable plan name.	required
description	str | None	What this plan covers.	None
test_cases_json	str | None	JSON array of test case objects. Each element must have at least {"code": "...", "name": "..."}. Additional fields: description, execution_mode ("once", "per_device", "per_binding").	None

Returns:

Type	Description
str	The test plan's registry string ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_test_plan(
    operation_id: str,
    code: str,
    name: str,
    description: str | None = None,
    test_cases_json: str | None = None,
) -> str:
    """Add a test plan to an Operation.

    A test plan groups related test cases.  Each test case defines a specific
    check to execute against a target (device, peripheral, cloud resource).

    Args:
        operation_id: Registry ID of the parent Operation.
        code: Unique plan code (e.g. ``"TP-UART-001"``).
        name: Human-readable plan name.
        description: What this plan covers.
        test_cases_json: JSON array of test case objects.  Each element must
                         have at least ``{"code": "...", "name": "..."}``.
                         Additional fields: ``description``, ``execution_mode``
                         (``"once"``, ``"per_device"``, ``"per_binding"``).

    Returns:
        The test plan's registry string ID.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    try:
        raw_cases: list[dict[str, Any]] = json.loads(test_cases_json or "[]")
    except json.JSONDecodeError:
        raw_cases = []
    cases = [TestCase(**tc) for tc in raw_cases] if raw_cases else None
    tp = TestPlan(code=code, name=name, description=description or "", test_cases=cases)
    op.addTestPlan(tp)
    tp_id = registry.store(tp, "test_plan", code, prefix="tp")
    return f"Test plan added. ID: {tp_id}"

add_test_plan_from_json(operation_id, json_path) async

Load a test plan from a JSON file and add it to an Operation.

Wintermute ships test plans in the TestPlans/ directory. Pass the path to one of those JSON files.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
json_path	str	Filesystem path to the test plan JSON file (e.g. "TestPlans/TP-HW-BLACKBOX-001.json").	required

Returns:

Type	Description
str	The test plan's registry string ID, or an error if the file is invalid.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_test_plan_from_json(
    operation_id: str,
    json_path: str,
) -> str:
    """Load a test plan from a JSON file and add it to an Operation.

    Wintermute ships test plans in the ``TestPlans/`` directory.  Pass the
    path to one of those JSON files.

    Args:
        operation_id: Registry ID of the parent Operation.
        json_path: Filesystem path to the test plan JSON file
                   (e.g. ``"TestPlans/TP-HW-BLACKBOX-001.json"``).

    Returns:
        The test plan's registry string ID, or an error if the file is invalid.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    try:
        with open(json_path, "r") as f:
            data: dict[str, Any] = json.load(f)
    except Exception as e:
        return f"Failed to read '{json_path}': {e}"
    tp = TestPlan.from_dict(data)
    op.addTestPlan(tp)
    tp_id = registry.store(tp, "test_plan", tp.code, prefix="tp")
    return f"Loaded test plan '{tp.name}' ({tp.code}). ID: {tp_id}"

add_user(operation_id, uid, name, email=None, teams_json=None, dept=None) async

Add an organizational user (employee, contractor) to an Operation.

Users represent people in the target organization whose access and permissions are under review.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
uid	str	Unique user identifier (e.g. "jsmith").	required
name	str	Full name (e.g. "John Smith").	required
email	str | None	Email address.	None
teams_json	str | None	JSON array of team names (e.g. '["red-team","infra"]').	None
dept	str | None	Department name.	None

Returns:

Type	Description
str	The user's registry string ID.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_user(
    operation_id: str,
    uid: str,
    name: str,
    email: str | None = None,
    teams_json: str | None = None,
    dept: str | None = None,
) -> str:
    """Add an organizational user (employee, contractor) to an Operation.

    Users represent people in the target organization whose access and
    permissions are under review.

    Args:
        operation_id: Registry ID of the parent Operation.
        uid: Unique user identifier (e.g. ``"jsmith"``).
        name: Full name (e.g. ``"John Smith"``).
        email: Email address.
        teams_json: JSON array of team names (e.g. ``'["red-team","infra"]'``).
        dept: Department name.

    Returns:
        The user's registry string ID.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    try:
        teams: list[str] = json.loads(teams_json or '["default"]')
    except json.JSONDecodeError:
        teams = ["default"]
    op.addUser(uid=uid, name=name, email=email or "", teams=teams, dept=dept or "")
    user = next((u for u in op.users if u.uid == uid), None)
    if user is None:
        return f"addUser returned success but user '{uid}' not found."
    uid_reg = registry.store(user, "user", uid, prefix="user")
    return f"User added. ID: {uid_reg}"

add_vulnerability_to_test_run(test_run_id, title, cvss, description='') async

Attach a new :class:Vulnerability directly to a TestCaseRun's findings list.

This is the run-level counterpart to :func:add_vulnerability (which attaches to a Service). Use it when the finding came out of a specific test execution and should travel with that run for reporting.

Parameters:

Name	Type	Description	Default
test_run_id	str	Registry ID of the target TestCaseRun.	required
title	str	Short human-readable label for the vulnerability.	required
cvss	int	CVSS score as an integer (the framework's existing field type — pass the rounded score).	required
description	str	Optional longer write-up of the issue.	''

Returns:

Type	Description
str	JSON object with run_id, vuln_id (auto-generated UUID),
str	and findings_count after the append, or {"error": "..."}.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def add_vulnerability_to_test_run(
    test_run_id: str,
    title: str,
    cvss: int,
    description: str = "",
) -> str:
    """Attach a new :class:`Vulnerability` directly to a TestCaseRun's
    ``findings`` list.

    This is the run-level counterpart to :func:`add_vulnerability` (which
    attaches to a Service). Use it when the finding came out of a specific
    test execution and should travel with that run for reporting.

    Args:
        test_run_id: Registry ID of the target TestCaseRun.
        title: Short human-readable label for the vulnerability.
        cvss: CVSS score as an integer (the framework's existing field
            type — pass the rounded score).
        description: Optional longer write-up of the issue.

    Returns:
        JSON object with ``run_id``, ``vuln_id`` (auto-generated UUID),
        and ``findings_count`` after the append, or ``{"error": "..."}``.
    """
    try:
        run: TestCaseRun = _require(test_run_id, TestCaseRun, "TestCaseRun")
    except (ValueError, TypeError) as exc:
        return json.dumps({"error": str(exc)}, indent=2)
    vuln = Vulnerability(title=title, cvss=cvss, description=description)
    run.findings.append(vuln)
    return json.dumps(
        {
            "run_id": run.run_id,
            "vuln_id": vuln.vuln_id,
            "findings_count": len(run.findings),
        },
        indent=2,
    )

ai_chat(prompt, task_tag=None) async

Send a one-shot prompt to the configured LLM and return the response.

You MUST call configure_ai_router() before using this tool.

Parameters:

Name	Type	Description	Default
prompt	str	The question or instruction for the LLM.	required
task_tag	str | None	Routing hint. Use "cheap" to route to Groq for fast inexpensive queries.	None

Returns:

Type	Description
str	The LLM's text response.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def ai_chat(prompt: str, task_tag: str | None = None) -> str:
    """Send a one-shot prompt to the configured LLM and return the response.

    You MUST call configure_ai_router() before using this tool.

    Args:
        prompt: The question or instruction for the LLM.
        task_tag: Routing hint.  Use ``"cheap"`` to route to Groq for fast
                  inexpensive queries.

    Returns:
        The LLM's text response.
    """
    router = _router()
    if router is None:
        return "AI router not initialized. Call configure_ai_router() first."
    try:
        return simple_chat(router, prompt, task_tag or "generic")
    except Exception as e:
        return f"AI chat failed: {e}"

analyze_test_coverage(operation_id) async

Analyze test coverage across all assets in an Operation.

Categorizes test runs by asset type (EC2, S3, IAM, Device, etc.) and returns a coverage breakdown.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required

Returns:

Type	Description
str	JSON object mapping asset categories to test run counts.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def analyze_test_coverage(operation_id: str) -> str:
    """Analyze test coverage across all assets in an Operation.

    Categorizes test runs by asset type (EC2, S3, IAM, Device, etc.) and
    returns a coverage breakdown.

    Args:
        operation_id: Registry ID of the parent Operation.

    Returns:
        JSON object mapping asset categories to test run counts.
    """
    from wintermute.utils.coverage import analyze_coverage

    op: Operation = _require(operation_id, Operation, "Operation")
    result = analyze_coverage(op)
    return json.dumps(result, indent=2)

attach_evidence(target_id, vuln_title, artifact_data, artifact_title=None, tool=None, action=None) async

Attach evidence (crash dump, token, PoC output) to a vulnerability.

This creates a ReproductionStep on the specified vulnerability with the artifact data stored in the vulnOutput field. Use this to log proof of exploitation or raw tool output.

Parameters:

Name	Type	Description	Default
target_id	str	Registry ID of the object holding the vulnerability.	required
vuln_title	str	Title of the vulnerability to attach evidence to.	required
artifact_data	str	The raw evidence content (text, base64, hex dump, etc.).	required
artifact_title	str | None	Short label for this evidence item.	None
tool	str | None	Tool that produced this evidence (e.g. "gdb", "nmap").	None
action	str | None	Action performed (e.g. "exploit", "dump").	None

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def attach_evidence(
    target_id: str,
    vuln_title: str,
    artifact_data: str,
    artifact_title: str | None = None,
    tool: str | None = None,
    action: str | None = None,
) -> str:
    """Attach evidence (crash dump, token, PoC output) to a vulnerability.

    This creates a ReproductionStep on the specified vulnerability with the
    artifact data stored in the ``vulnOutput`` field.  Use this to log proof
    of exploitation or raw tool output.

    Args:
        target_id: Registry ID of the object holding the vulnerability.
        vuln_title: Title of the vulnerability to attach evidence to.
        artifact_data: The raw evidence content (text, base64, hex dump, etc.).
        artifact_title: Short label for this evidence item.
        tool: Tool that produced this evidence (e.g. ``"gdb"``, ``"nmap"``).
        action: Action performed (e.g. ``"exploit"``, ``"dump"``).

    Returns:
        Confirmation or error message.
    """
    obj = registry.get(target_id)
    if obj is None:
        return f"ID '{target_id}' not found."
    step = ReproductionStep(
        title=artifact_title or "evidence",
        description="Evidence attached by agent",
        tool=tool or None,
        action=action or None,
        confidence=100,
        vulnOutput=artifact_data,
    )
    result = add_reproduction_step(obj, title=vuln_title, step=step)
    if result:
        return f"Evidence '{artifact_title}' attached to vulnerability '{vuln_title}'."
    vuln = get_vulnerability(obj, title=vuln_title)
    if vuln is None:
        return f"Vulnerability '{vuln_title}' not found on {target_id}."
    vuln.reproduction_steps.append(step)
    return f"Evidence '{artifact_title}' attached to vulnerability '{vuln_title}'."

close_ssh_session(session_id) async

Close a persistent SSH session and release its resources.

Parameters:

Name	Type	Description	Default
session_id	str	ID returned by open_ssh_session.	required

Returns:

Type	Description
str	Confirmation message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def close_ssh_session(session_id: str) -> str:
    """Close a persistent SSH session and release its resources.

    Args:
        session_id: ID returned by ``open_ssh_session``.

    Returns:
        Confirmation message.
    """
    from wintermute.ai.utils.ssh_exec import SSHSession

    try:
        session: SSHSession = _require(session_id, SSHSession, "SSHSession")
        await session.close()
        registry.delete(session_id)
        return json.dumps({"result": f"Session {session_id} closed."}, indent=2)
    except (ValueError, TypeError) as e:
        return json.dumps({"error": str(e)}, indent=2)

configure_ai_router() async

Initialize the Wintermute AI router and register all LLM providers.

This bootstraps Bedrock, Groq, OpenAI, and HuggingFace providers using environment variables (AWS_REGION, GROQ_API_KEY, OPENAI_API_KEY, BEDROCK_MODEL_ID). It also scans ./knowledge_bases/ for RAG indices and registers them as rag-<name> providers.

Run this tool ONCE at the start of any session that requires AI features (enrichment, chat, tool calling).

Returns:

Type	Description
str	List of registered providers and the default provider name.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def configure_ai_router() -> str:
    """Initialize the Wintermute AI router and register all LLM providers.

    This bootstraps Bedrock, Groq, OpenAI, and HuggingFace providers using
    environment variables (``AWS_REGION``, ``GROQ_API_KEY``, ``OPENAI_API_KEY``,
    ``BEDROCK_MODEL_ID``).  It also scans ``./knowledge_bases/`` for RAG
    indices and registers them as ``rag-<name>`` providers.

    Run this tool ONCE at the start of any session that requires AI features
    (enrichment, chat, tool calling).

    Returns:
        List of registered providers and the default provider name.
    """
    global _ai_router  # noqa: PLW0603
    try:
        _ai_router = init_router()
        return json.dumps(
            {
                "status": "ok",
                "default_provider": _ai_router.default_provider,
                "available_providers": llms.get_provider_descriptions(),
            },
            indent=2,
        )
    except Exception as e:
        return f"Router init failed: {e}"

create_operation(name, ticket=None, start_date=None, end_date=None) async

Create a new top-level Wintermute Operation and register it.

An Operation is the root container for an entire engagement — all devices, users, cloud accounts, test plans, and findings live beneath it. You MUST create an Operation before adding any other objects.

Parameters:

Name	Type	Description	Default
name	str	A short, unique identifier for the engagement (e.g. "acme-iot-audit-2025").	required
ticket	str | None	Optional ticket or tracking reference.	None
start_date	str | None	Engagement start in MM/DD/YYYY format (defaults to today).	None
end_date	str | None	Engagement end in MM/DD/YYYY format (defaults to today).	None

Returns:

Type	Description
str	The string ID assigned to this Operation (e.g. "op:acme-iot-audit-2025").
str	Use this ID in all subsequent calls that require an operation_id.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def create_operation(
    name: str,
    ticket: str | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
) -> str:
    """Create a new top-level Wintermute Operation and register it.

    An Operation is the root container for an entire engagement — all devices,
    users, cloud accounts, test plans, and findings live beneath it.  You MUST
    create an Operation before adding any other objects.

    Args:
        name: A short, unique identifier for the engagement (e.g.
              ``"acme-iot-audit-2025"``).
        ticket: Optional ticket or tracking reference.
        start_date: Engagement start in MM/DD/YYYY format (defaults to today).
        end_date: Engagement end in MM/DD/YYYY format (defaults to today).

    Returns:
        The string ID assigned to this Operation (e.g. ``"op:acme-iot-audit-2025"``).
        Use this ID in all subsequent calls that require an ``operation_id``.
    """
    kwargs: dict[str, Any] = {"operation_name": name}
    if ticket:
        kwargs["ticket"] = ticket
    if start_date:
        kwargs["start_date"] = start_date
    if end_date:
        kwargs["end_date"] = end_date
    op = Operation(**kwargs)
    oid = registry.store(op, "operation", name, prefix="op")
    return f"Created Operation. ID: {oid}"

create_ticket(title, description=None, assignee=None, requester=None) async

Create a new ticket in the configured ticketing backend.

You MUST call setup_ticket_backend() first.

Parameters:

Name	Type	Description	Default
title	str	Short summary of the issue.	required
description	str | None	Detailed description or reproduction notes.	None
assignee	str | None	Person assigned to the ticket.	None
requester	str | None	Person who requested the ticket.	None

Returns:

Type	Description
str	The ticket ID string assigned by the backend.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def create_ticket(
    title: str,
    description: str | None = None,
    assignee: str | None = None,
    requester: str | None = None,
) -> str:
    """Create a new ticket in the configured ticketing backend.

    You MUST call setup_ticket_backend() first.

    Args:
        title: Short summary of the issue.
        description: Detailed description or reproduction notes.
        assignee: Person assigned to the ticket.
        requester: Person who requested the ticket.

    Returns:
        The ticket ID string assigned by the backend.
    """
    try:
        kwargs: dict[str, Any] = {"title": title, "description": description or ""}
        if assignee:
            kwargs["assignee"] = assignee
        if requester:
            kwargs["requester"] = requester
        tid: str = Ticket.create(**kwargs)
        return f"Ticket created: {tid}"
    except Exception as e:
        return f"Failed: {e}"

delete_analyst(operation_id, userid) async

Remove an analyst from an Operation by userid.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
userid	str	The analyst's userid to remove.	required

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def delete_analyst(operation_id: str, userid: str) -> str:
    """Remove an analyst from an Operation by userid.

    Args:
        operation_id: Registry ID of the parent Operation.
        userid: The analyst's userid to remove.

    Returns:
        Confirmation or error message.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    result = op.delAnalyst(userid)
    return f"Analyst '{userid}' removed: {result}"

delete_cloud_account(operation_id, name, cloud_account_id=None) async

Remove a cloud account from an Operation and the registry.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
name	str	The name field of the cloud account to remove.	required
cloud_account_id	str | None	Registry ID to un-register (optional).	None

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def delete_cloud_account(
    operation_id: str,
    name: str,
    cloud_account_id: str | None = None,
) -> str:
    """Remove a cloud account from an Operation and the registry.

    Args:
        operation_id: Registry ID of the parent Operation.
        name: The ``name`` field of the cloud account to remove.
        cloud_account_id: Registry ID to un-register (optional).

    Returns:
        Confirmation or error message.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    result = op.delCloudAccount(name)
    if cloud_account_id:
        registry.delete(cloud_account_id)
    return f"Cloud account '{name}' removed: {result}"

delete_device(operation_id, hostname=None, device_id=None) async

Remove a Device from an Operation and the registry.

Provide at least one of hostname or device_id to identify the device. If both are given, both are used.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
hostname	str | None	Hostname of the device to remove.	None
device_id	str | None	Registry ID of the device to un-register.	None

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def delete_device(
    operation_id: str,
    hostname: str | None = None,
    device_id: str | None = None,
) -> str:
    """Remove a Device from an Operation and the registry.

    Provide at least one of ``hostname`` or ``device_id`` to identify
    the device.  If both are given, both are used.

    Args:
        operation_id: Registry ID of the parent Operation.
        hostname: Hostname of the device to remove.
        device_id: Registry ID of the device to un-register.

    Returns:
        Confirmation or error message.
    """
    if not hostname and not device_id:
        return "Provide at least one of 'hostname' or 'device_id'."
    op: Operation = _require(operation_id, Operation, "Operation")
    result_parts: list[str] = []
    if hostname:
        removed = op.delDevice(hostname)
        result_parts.append(f"Device '{hostname}' removed from operation: {removed}")
    if device_id:
        registry.delete(device_id)
        result_parts.append(f"Registry entry '{device_id}' deleted.")
    return " ".join(result_parts)

delete_operation(operation_id) async

Remove an Operation and all its child references from the registry.

WARNING: This deletes the in-memory Operation object. Any devices, users, or cloud accounts that were registered from this Operation will become orphaned in the registry. Call list_active_objects() afterward to clean up.

Parameters:

Name	Type	Description	Default
operation_id	str	The registry string ID of the Operation to delete.	required

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def delete_operation(operation_id: str) -> str:
    """Remove an Operation and all its child references from the registry.

    WARNING: This deletes the in-memory Operation object. Any devices, users,
    or cloud accounts that were registered from this Operation will become
    orphaned in the registry.  Call list_active_objects() afterward to clean up.

    Args:
        operation_id: The registry string ID of the Operation to delete.

    Returns:
        Confirmation or error message.
    """
    if registry.delete(operation_id):
        return f"Deleted {operation_id} from registry."
    return f"ID '{operation_id}' not found."

delete_user(operation_id, uid, user_id=None) async

Remove a user from an Operation.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
uid	str	The user's uid field to delete from the Operation.	required
user_id	str | None	Registry ID of the user to un-register (optional).	None

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def delete_user(operation_id: str, uid: str, user_id: str | None = None) -> str:
    """Remove a user from an Operation.

    Args:
        operation_id: Registry ID of the parent Operation.
        uid: The user's uid field to delete from the Operation.
        user_id: Registry ID of the user to un-register (optional).

    Returns:
        Confirmation or error message.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    result = op.delUser(uid)
    if user_id:
        registry.delete(user_id)
    return f"User '{uid}' removed: {result}"

download_file_ssh(target_alias, remote_path, local_path, username=None, password=None, port=None) async

Download a file from a remote host via SFTP/SSH.

Uses asyncssh and the host machine's ~/.ssh/config for proxy jumping, key management, and host aliases.

Parameters:

Name	Type	Description	Default
target_alias	str	SSH host alias from ~/.ssh/config, or hostname/IP.	required
remote_path	str	File path on the remote host to download.	required
local_path	str	Local destination path for the downloaded file.	required
username	str | None	SSH username (optional — inferred from ssh config).	None
password	str | None	SSH password (optional — falls back to key-based auth).	None
port	int | None	SSH port (optional — defaults to ssh config or 22).	None

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def download_file_ssh(
    target_alias: str,
    remote_path: str,
    local_path: str,
    username: str | None = None,
    password: str | None = None,
    port: int | None = None,
) -> str:
    """Download a file from a remote host via SFTP/SSH.

    Uses asyncssh and the host machine's ``~/.ssh/config`` for proxy jumping,
    key management, and host aliases.

    Args:
        target_alias: SSH host alias from ~/.ssh/config, or hostname/IP.
        remote_path: File path on the remote host to download.
        local_path: Local destination path for the downloaded file.
        username: SSH username (optional — inferred from ssh config).
        password: SSH password (optional — falls back to key-based auth).
        port: SSH port (optional — defaults to ssh config or 22).

    Returns:
        Confirmation or error message.
    """
    from wintermute.ai.utils.ssh_exec import download_file_async

    result = await download_file_async(
        target_alias=target_alias,
        remote_path=remote_path,
        local_path=local_path,
        username=username,
        password=password,
        port=port,
    )
    return json.dumps(result, indent=2, default=str)

download_file_ssh_session(session_id, remote_path, local_path) async

Download a file via SFTP on a persistent SSH session.

Parameters:

Name	Type	Description	Default
session_id	str	ID returned by open_ssh_session.	required
remote_path	str	File path on the remote host to download.	required
local_path	str	Local destination path for the downloaded file.	required

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def download_file_ssh_session(
    session_id: str, remote_path: str, local_path: str
) -> str:
    """Download a file via SFTP on a persistent SSH session.

    Args:
        session_id: ID returned by ``open_ssh_session``.
        remote_path: File path on the remote host to download.
        local_path: Local destination path for the downloaded file.

    Returns:
        Confirmation or error message.
    """
    from wintermute.ai.utils.ssh_exec import SSHSession

    try:
        session: SSHSession = _require(session_id, SSHSession, "SSHSession")
        result = await session.download(remote_path, local_path)
        return json.dumps(result, indent=2, default=str)
    except (ValueError, TypeError) as e:
        return json.dumps({"error": str(e)}, indent=2)

edit_aws_account(aws_account_id, description=None, default_region=None) async

Modify mutable fields on a registered AWS Account.

Parameters:

Name	Type	Description	Default
aws_account_id	str	Registry ID of the target AWSAccount.	required
description	str | None	New description (empty to skip).	None
default_region	str | None	New default region (empty to skip).	None

Returns:

Type	Description
str	Confirmation with changed fields.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def edit_aws_account(
    aws_account_id: str,
    description: str | None = None,
    default_region: str | None = None,
) -> str:
    """Modify mutable fields on a registered AWS Account.

    Args:
        aws_account_id: Registry ID of the target AWSAccount.
        description: New description (empty to skip).
        default_region: New default region (empty to skip).

    Returns:
        Confirmation with changed fields.
    """
    acc: AWSAccount = _require(aws_account_id, AWSAccount, "AWSAccount")
    changed: list[str] = []
    if description:
        acc.description = description
        changed.append("description")
    if default_region:
        acc.default_region = default_region
        changed.append("default_region")
    if not changed:
        return "No fields provided."
    return f"Updated {aws_account_id}: {', '.join(changed)}"

edit_device(device_id, operating_system=None, fqdn=None, macaddr=None) async

Modify mutable fields on a registered Device.

Parameters:

Name	Type	Description	Default
device_id	str	Registry ID of the target Device.	required
operating_system	str | None	New OS string (empty to skip).	None
fqdn	str | None	New FQDN (empty to skip).	None
macaddr	str | None	New MAC address (empty to skip).	None

Returns:

Type	Description
str	Confirmation with changed fields.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def edit_device(
    device_id: str,
    operating_system: str | None = None,
    fqdn: str | None = None,
    macaddr: str | None = None,
) -> str:
    """Modify mutable fields on a registered Device.

    Args:
        device_id: Registry ID of the target Device.
        operating_system: New OS string (empty to skip).
        fqdn: New FQDN (empty to skip).
        macaddr: New MAC address (empty to skip).

    Returns:
        Confirmation with changed fields.
    """
    from wintermute.core import Device

    dev: Device = _require(device_id, Device, "Device")
    changed: list[str] = []
    if operating_system:
        dev.operatingsystem = operating_system
        changed.append("operatingsystem")
    if fqdn:
        dev.fqdn = fqdn
        changed.append("fqdn")
    if macaddr:
        dev.macaddr = macaddr
        changed.append("macaddr")
    if not changed:
        return "No fields provided."
    return f"Updated {device_id}: {', '.join(changed)}"

edit_operation(operation_id, ticket=None, start_date=None, end_date=None) async

Modify mutable fields on an existing Operation.

You MUST supply a valid operation_id obtained from create_operation().

Parameters:

Name	Type	Description	Default
operation_id	str	The registry string ID of the target Operation.	required
ticket	str | None	New ticket reference (leave empty to keep current value).	None
start_date	str | None	New start date in MM/DD/YYYY (leave empty to keep current).	None
end_date	str | None	New end date in MM/DD/YYYY (leave empty to keep current).	None

Returns:

Type	Description
str	Confirmation message with the updated fields.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def edit_operation(
    operation_id: str,
    ticket: str | None = None,
    start_date: str | None = None,
    end_date: str | None = None,
) -> str:
    """Modify mutable fields on an existing Operation.

    You MUST supply a valid ``operation_id`` obtained from create_operation().

    Args:
        operation_id: The registry string ID of the target Operation.
        ticket: New ticket reference (leave empty to keep current value).
        start_date: New start date in MM/DD/YYYY (leave empty to keep current).
        end_date: New end date in MM/DD/YYYY (leave empty to keep current).

    Returns:
        Confirmation message with the updated fields.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    changed: list[str] = []
    if ticket:
        op.ticket = ticket
        changed.append("ticket")
    if start_date:
        op.start_date = start_date
        changed.append("start_date")
    if end_date:
        op.end_date = end_date
        changed.append("end_date")
    if not changed:
        return "No fields provided — nothing changed."
    return f"Updated {operation_id}: {', '.join(changed)}"

enrich_processor_via_ai(processor_name, manufacturer=None) async

Use AI to enrich a processor with architecture and capability details.

Creates a minimal Processor object and queries the LLM to fill in ISA, core count, security features, pinout, and known vulnerabilities.

You MUST call configure_ai_router() first.

Parameters:

Name	Type	Description	Default
processor_name	str	Processor name/model (e.g. "STM32F407").	required
manufacturer	str | None	Manufacturer name (e.g. "STMicroelectronics").	None

Returns:

Type	Description
str	JSON representation of the enriched Processor.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def enrich_processor_via_ai(
    processor_name: str,
    manufacturer: str | None = None,
) -> str:
    """Use AI to enrich a processor with architecture and capability details.

    Creates a minimal Processor object and queries the LLM to fill in ISA,
    core count, security features, pinout, and known vulnerabilities.

    You MUST call configure_ai_router() first.

    Args:
        processor_name: Processor name/model (e.g. ``"STM32F407"``).
        manufacturer: Manufacturer name (e.g. ``"STMicroelectronics"``).

    Returns:
        JSON representation of the enriched Processor.
    """
    from wintermute.ai.utils.hardware import enrich_processor

    proc = Processor(processor=processor_name, manufacturer=manufacturer or None)
    try:
        enriched = enrich_processor(proc, router=_router())
        pid = registry.store(enriched, "processor", processor_name, prefix="proc")
        result = enriched.to_dict()
        result["_registry_id"] = pid
        return json.dumps(result, indent=2, default=str)
    except Exception as e:
        return f"Enrichment failed: {e}"

execute_depthcharge_catalog(peripheral_id, add_vulns=None) async

Run Depthcharge command cataloging against a UART peripheral.

Connects to the U-Boot console via the peripheral's device_path, enumerates all available commands, scores them for danger, and optionally adds vulnerabilities for dangerous commands.

REQUIRES: Physical access to the target device's UART console running U-Boot, and the depthcharge Python package.

Parameters:

Name	Type	Description	Default
peripheral_id	str	Registry ID of the target UART Peripheral.	required
add_vulns	bool | None	If True, automatically create vulnerabilities for dangerous commands found.	None

Returns:

Type	Description
str	JSON summary of discovered commands and danger analysis.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def execute_depthcharge_catalog(
    peripheral_id: str,
    add_vulns: bool | None = None,
) -> str:
    """Run Depthcharge command cataloging against a UART peripheral.

    Connects to the U-Boot console via the peripheral's device_path, enumerates
    all available commands, scores them for danger, and optionally adds
    vulnerabilities for dangerous commands.

    REQUIRES: Physical access to the target device's UART console running
    U-Boot, and the ``depthcharge`` Python package.

    Args:
        peripheral_id: Registry ID of the target UART Peripheral.
        add_vulns: If True, automatically create vulnerabilities for dangerous
                   commands found.

    Returns:
        JSON summary of discovered commands and danger analysis.
    """
    from wintermute.backends.depthcharge import DepthchargePeripheralAgent
    from wintermute.basemodels import Peripheral

    periph: Peripheral = _require(peripheral_id, Peripheral, "Peripheral")
    try:
        # Create a memory buffer to swallow the rogue print statements
        f = io.StringIO()

        # Everything inside this 'with' block has its terminal output silenced
        with redirect_stdout(f), redirect_stderr(f):
            agent = DepthchargePeripheralAgent(peripheral=periph)  # type: ignore[arg-type]
            result = agent.catalog_commands_and_flag(
                addVulns=add_vulns if add_vulns is not None else True
            )

        # Once we exit the 'with' block, standard output is restored safely!
        return json.dumps(result, indent=2, default=str)
    except Exception as e:
        return f"Depthcharge catalog failed: {e}"

execute_depthcharge_memory_dump(peripheral_id, address, length, filename=None) async

Dump memory from a U-Boot target via Depthcharge.

Connects to the U-Boot console and reads raw memory. The dump is saved to disk and a vulnerability is automatically created if successful.

REQUIRES: Physical access to the target device's UART console running U-Boot.

Parameters:

Name	Type	Description	Default
peripheral_id	str	Registry ID of the target UART Peripheral.	required
address	str	Start address as a hex string (e.g. "0x80000000").	required
length	int	Number of bytes to dump.	required
filename	str | None	Output filename (auto-generated if empty).	None

Returns:

Type	Description
str	JSON summary with file path and SHA256 hash.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def execute_depthcharge_memory_dump(
    peripheral_id: str,
    address: str,
    length: int,
    filename: str | None = None,
) -> str:
    """Dump memory from a U-Boot target via Depthcharge.

    Connects to the U-Boot console and reads raw memory.  The dump is saved
    to disk and a vulnerability is automatically created if successful.

    REQUIRES: Physical access to the target device's UART console running
    U-Boot.

    Args:
        peripheral_id: Registry ID of the target UART Peripheral.
        address: Start address as a hex string (e.g. ``"0x80000000"``).
        length: Number of bytes to dump.
        filename: Output filename (auto-generated if empty).

    Returns:
        JSON summary with file path and SHA256 hash.
    """
    from wintermute.backends.depthcharge import DepthchargePeripheralAgent
    from wintermute.basemodels import Peripheral

    periph: Peripheral = _require(peripheral_id, Peripheral, "Peripheral")
    try:
        agent = DepthchargePeripheralAgent(peripheral=periph)  # type: ignore[arg-type]
        addr_int = int(address, 16) if address.startswith("0x") else int(address)
        result = agent.dump_memory_and_attach_vuln(
            address=addr_int,
            length=length,
            filename=filename or None,
        )
        return json.dumps(result, indent=2, default=str)
    except Exception as e:
        return f"Memory dump failed: {e}"

execute_tpm20_get_random(device_path=None, num_bytes=None) async

Execute the TPM 2.0 GetRandom command via the tpm20 cartridge.

Requires the tpm20 cartridge to be loaded and a TPM device accessible at device_path.

Parameters:

Name	Type	Description	Default
device_path	str | None	Path to the TPM device (default "/dev/tpm0").	None
num_bytes	int | None	Number of random bytes to request (1-64).	None

Returns:

Type	Description
str	Hex-encoded random bytes or error.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def execute_tpm20_get_random(
    device_path: str | None = None,
    num_bytes: int | None = None,
) -> str:
    """Execute the TPM 2.0 GetRandom command via the tpm20 cartridge.

    Requires the ``tpm20`` cartridge to be loaded and a TPM device accessible
    at ``device_path``.

    Args:
        device_path: Path to the TPM device (default ``"/dev/tpm0"``).
        num_bytes: Number of random bytes to request (1-64).

    Returns:
        Hex-encoded random bytes or error.
    """
    if "tpm20" not in _loaded_cartridges:
        return "Cartridge 'tpm20' not loaded. Call load_cartridge('tpm20') first."
    mod = _loaded_cartridges["tpm20"]
    try:
        transport = mod.TPMTransport(device_path=device_path or "/dev/tpm0")
        tpm = mod.tpm20(transport=transport)
        data: bytes = tpm.get_random(num_bytes if num_bytes is not None else 16)
        return f"Random bytes ({len(data)}): {data.hex()}"
    except Exception as e:
        return f"TPM GetRandom failed: {e}"

generate_report(operation_id, title=None, output_path=None, author=None, summary=None) async

Generate a DOCX report from an Operation's findings.

You MUST call setup_report_backend() first. The report collects all vulnerabilities found across the Operation's devices, services, and cloud accounts and renders them using the configured templates.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the source Operation.	required
title	str | None	Report title (appears on the cover page).	None
output_path	str | None	Filesystem path where the .docx file will be saved.	None
author	str | None	Report author name.	None
summary	str | None	Executive summary paragraph.	None

Returns:

Type	Description
str	Confirmation with the output path.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def generate_report(
    operation_id: str,
    title: str | None = None,
    output_path: str | None = None,
    author: str | None = None,
    summary: str | None = None,
) -> str:
    """Generate a DOCX report from an Operation's findings.

    You MUST call setup_report_backend() first.  The report collects all
    vulnerabilities found across the Operation's devices, services, and cloud
    accounts and renders them using the configured templates.

    Args:
        operation_id: Registry ID of the source Operation.
        title: Report title (appears on the cover page).
        output_path: Filesystem path where the .docx file will be saved.
        author: Report author name.
        summary: Executive summary paragraph.

    Returns:
        Confirmation with the output path.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    title = title or "Pentest Report"
    output_path = output_path or "report_output.docx"
    spec = ReportSpec(title=title, author=author or None, summary=summary or "")
    try:
        Report.save(spec, [op], output_path)
        return f"Report saved to '{output_path}'."
    except Exception as e:
        return f"Report generation failed: {e}"

generate_strategy_report(operation_id) async

Use AI to generate an execution strategy report for an Operation.

Analyzes all test runs, findings, and targets in the Operation and produces a markdown-formatted strategy document.

You MUST call configure_ai_router() first.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required

Returns:

Type	Description
str	Markdown-formatted strategy report text.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def generate_strategy_report(operation_id: str) -> str:
    """Use AI to generate an execution strategy report for an Operation.

    Analyzes all test runs, findings, and targets in the Operation and
    produces a markdown-formatted strategy document.

    You MUST call configure_ai_router() first.

    Args:
        operation_id: Registry ID of the parent Operation.

    Returns:
        Markdown-formatted strategy report text.
    """
    from wintermute.ai.reporting import generate_execution_strategy_report

    router = _router()
    if router is None:
        return "AI router not initialized. Call configure_ai_router() first."
    op: Operation = _require(operation_id, Operation, "Operation")
    try:
        return generate_execution_strategy_report(router, op)
    except Exception as e:
        return f"Strategy report failed: {e}"

generate_test_runs(operation_id, replace=None) async

Generate test case runs from all test plans in an Operation.

This expands every test case across its execution mode (once, per_device, per_binding) and creates TestCaseRun entries with status not_run.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required
replace	bool | None	If True, clear existing runs before generating new ones.	None

Returns:

Type	Description
str	JSON list of generated run IDs and their test case codes.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def generate_test_runs(
    operation_id: str,
    replace: bool | None = None,
) -> str:
    """Generate test case runs from all test plans in an Operation.

    This expands every test case across its execution mode (once, per_device,
    per_binding) and creates ``TestCaseRun`` entries with status ``not_run``.

    Args:
        operation_id: Registry ID of the parent Operation.
        replace: If ``True``, clear existing runs before generating new ones.

    Returns:
        JSON list of generated run IDs and their test case codes.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    runs = op.generateTestRuns(replace=replace if replace is not None else False)
    result: list[dict[str, str]] = []
    for run in runs:
        rid = registry.store(run, "test_run", run.run_id[:8], prefix="run")
        result.append(
            {
                "registry_id": rid,
                "run_id": run.run_id,
                "test_case_code": run.test_case_code,
            }
        )
    return json.dumps(result, indent=2)

get_cloud_account_info(cloud_account_id) async

Return all known properties of a Cloud/AWS Account.

Includes IAM users, IAM roles, services, tags, and attached vulnerabilities. Use this to map the attack surface of a cloud account.

Parameters:

Name	Type	Description	Default
cloud_account_id	str	Registry ID of the target Cloud/AWS Account.	required

Returns:

Type	Description
str	Full JSON representation of the account.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def get_cloud_account_info(cloud_account_id: str) -> str:
    """Return all known properties of a Cloud/AWS Account.

    Includes IAM users, IAM roles, services, tags, and attached
    vulnerabilities.  Use this to map the attack surface of a cloud account.

    Args:
        cloud_account_id: Registry ID of the target Cloud/AWS Account.

    Returns:
        Full JSON representation of the account.
    """
    acc = registry.get(cloud_account_id)
    if acc is None:
        return f"ID '{cloud_account_id}' not found."
    return _ser(acc)

get_device_info(device_id) async

Return all known properties of a Device including services and peripherals.

Use this to see open ports, JTAG/UART configurations, processor details, and attached vulnerabilities before deciding what to exploit or test.

Parameters:

Name	Type	Description	Default
device_id	str	Registry ID of the target Device.	required

Returns:

Type	Description
str	Full JSON representation of the Device.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def get_device_info(device_id: str) -> str:
    """Return all known properties of a Device including services and peripherals.

    Use this to see open ports, JTAG/UART configurations, processor details,
    and attached vulnerabilities before deciding what to exploit or test.

    Args:
        device_id: Registry ID of the target Device.

    Returns:
        Full JSON representation of the Device.
    """
    from wintermute.core import Device

    dev: Device = _require(device_id, Device, "Device")
    return _ser(dev)

get_operation_info(operation_id) async

Retrieve the full serialized state of an Operation.

Use this tool to inspect all devices, users, cloud accounts, test plans, test runs, and analysts that belong to an Operation. The response is a JSON document produced by Operation.to_dict().

Parameters:

Name	Type	Description	Default
operation_id	str	The registry string ID of the target Operation.	required

Returns:

Type	Description
str	Full JSON representation of the Operation.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def get_operation_info(operation_id: str) -> str:
    """Retrieve the full serialized state of an Operation.

    Use this tool to inspect all devices, users, cloud accounts, test plans,
    test runs, and analysts that belong to an Operation.  The response is a
    JSON document produced by ``Operation.to_dict()``.

    Args:
        operation_id: The registry string ID of the target Operation.

    Returns:
        Full JSON representation of the Operation.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    return _ser(op)

get_test_execution_status(operation_id) async

Generate a status report of all test runs in an Operation.

Returns aggregate counts by status (not_run, in_progress, passed, failed, blocked, not_applicable) for all test case runs.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the parent Operation.	required

Returns:

Type	Description
str	JSON status report with total_runs and by_status breakdown.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def get_test_execution_status(operation_id: str) -> str:
    """Generate a status report of all test runs in an Operation.

    Returns aggregate counts by status (not_run, in_progress, passed, failed,
    blocked, not_applicable) for all test case runs.

    Args:
        operation_id: Registry ID of the parent Operation.

    Returns:
        JSON status report with ``total_runs`` and ``by_status`` breakdown.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    now = datetime.now(timezone.utc)
    report = op.statusReport(start=now - timedelta(days=365), end=now)
    return json.dumps(report, indent=2, default=str)

get_test_plan_info(test_plan_id) async

Return the full content of a test plan including all test cases.

Parameters:

Name	Type	Description	Default
test_plan_id	str	Registry ID of the target TestPlan.	required

Returns:

Type	Description
str	Full JSON representation of the TestPlan.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def get_test_plan_info(test_plan_id: str) -> str:
    """Return the full content of a test plan including all test cases.

    Args:
        test_plan_id: Registry ID of the target TestPlan.

    Returns:
        Full JSON representation of the TestPlan.
    """
    tp: TestPlan = _require(test_plan_id, TestPlan, "TestPlan")
    return _ser(tp)

get_test_run_info(test_run_id) async

Return the current state of a test case run.

Includes status, timestamps, bound targets, findings, and notes.

Parameters:

Name	Type	Description	Default
test_run_id	str	Registry ID of the TestCaseRun.	required

Returns:

Type	Description
str	Full JSON representation of the TestCaseRun.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def get_test_run_info(test_run_id: str) -> str:
    """Return the current state of a test case run.

    Includes status, timestamps, bound targets, findings, and notes.

    Args:
        test_run_id: Registry ID of the TestCaseRun.

    Returns:
        Full JSON representation of the TestCaseRun.
    """
    run: TestCaseRun = _require(test_run_id, TestCaseRun, "TestCaseRun")
    return _ser(run)

get_user_info(user_id) async

Return all known properties of a User including teams and permissions.

Parameters:

Name	Type	Description	Default
user_id	str	Registry ID of the target User.	required

Returns:

Type	Description
str	Full JSON representation of the User.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def get_user_info(user_id: str) -> str:
    """Return all known properties of a User including teams and permissions.

    Args:
        user_id: Registry ID of the target User.

    Returns:
        Full JSON representation of the User.
    """
    from wintermute.core import User

    user: User = _require(user_id, User, "User")
    return _ser(user)

ingest_burp_scan(operation_id, burp_xml_path) async

Parse a Burp Suite XML export and add discovered hosts to an Operation.

The parser extracts Devices, Services, and Vulnerabilities from the Burp XML format and registers them in both the Operation and the session registry.

Parameters:

Name	Type	Description	Default
operation_id	str	Registry ID of the target Operation.	required
burp_xml_path	str	Filesystem path to the Burp XML file.	required

Returns:

Type	Description
str	Summary of ingested devices, services, and vulnerabilities.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def ingest_burp_scan(
    operation_id: str,
    burp_xml_path: str,
) -> str:
    """Parse a Burp Suite XML export and add discovered hosts to an Operation.

    The parser extracts Devices, Services, and Vulnerabilities from the Burp
    XML format and registers them in both the Operation and the session
    registry.

    Args:
        operation_id: Registry ID of the target Operation.
        burp_xml_path: Filesystem path to the Burp XML file.

    Returns:
        Summary of ingested devices, services, and vulnerabilities.
    """
    from wintermute.utils.parsers import BurpParser

    op: Operation = _require(operation_id, Operation, "Operation")
    parser = BurpParser()
    try:
        devices = parser.parse(file=burp_xml_path)
    except Exception as e:
        return f"Burp parse failed: {e}"

    dev_ids: list[str] = []
    total_svcs = 0
    total_vulns = 0
    for dev in devices:
        # Merge into operation — addDevice handles merge-on-hostname
        op.addDevice(
            dev.hostname, str(dev.ipaddr), dev.macaddr, dev.operatingsystem, dev.fqdn
        )
        existing = op.getDeviceByHostname(dev.hostname)
        if existing is None:
            continue
        # Merge services
        for svc in dev.services:
            existing.addService(
                protocol=svc.protocol,
                app=svc.app,
                portNumber=svc.portNumber,
                banner=svc.banner,
                transport_layer=svc.transport_layer,
            )
            total_svcs += 1
            for v in svc.vulnerabilities:
                existing.services[-1].addVulnerability(
                    title=v.title,
                    description=v.description,
                    threat=v.threat,
                    cvss=v.cvss,
                )
                total_vulns += 1
        did = registry.store(existing, "device", dev.hostname, prefix="dev")
        dev_ids.append(did)

    return json.dumps(
        {
            "devices_ingested": len(dev_ids),
            "services_ingested": total_svcs,
            "vulnerabilities_ingested": total_vulns,
            "device_ids": dev_ids,
        },
        indent=2,
    )

list_active_objects() async

Return every object currently held in the Wintermute session registry.

Call this tool FIRST in any new session to discover which string IDs are already available. The response is a JSON object mapping each string ID to its type label (e.g. "op:acme" → "operation").

Returns:

Type	Description
str	JSON mapping {string_id: type_label}.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def list_active_objects() -> str:
    """Return every object currently held in the Wintermute session registry.

    Call this tool FIRST in any new session to discover which string IDs are
    already available.  The response is a JSON object mapping each string ID
    to its type label (e.g. ``"op:acme" → "operation"``).

    Returns:
        JSON mapping ``{string_id: type_label}``.
    """
    data = registry.list_all()
    if not data:
        return "Registry is empty. Use create_operation() to begin."
    return json.dumps(data, indent=2)

list_cartridges() async

List all available Wintermute cartridge plugins.

Cartridges are located in wintermute/cartridges/. Each cartridge provides specialized hardware or offensive testing capabilities.

Returns:

Type	Description
str	JSON object with available and currently loaded cartridge names.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def list_cartridges() -> str:
    """List all available Wintermute cartridge plugins.

    Cartridges are located in ``wintermute/cartridges/``.  Each cartridge
    provides specialized hardware or offensive testing capabilities.

    Returns:
        JSON object with available and currently loaded cartridge names.
    """
    import pkgutil

    import wintermute.cartridges as pkg

    available: list[str] = []
    for _importer, modname, _ispkg in pkgutil.iter_modules(pkg.__path__):
        if not modname.startswith("_"):
            available.append(modname)
    return json.dumps(
        {"available": available, "loaded": list(_loaded_cartridges.keys())},
        indent=2,
    )

list_registered_ai_tools() async

List all tools currently registered in the Wintermute AI tool registry.

These are the tools available for LLM function-calling via tool_calling_chat().

Returns:

Type	Description
str	JSON array of tool definitions (name, description, parameters).

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def list_registered_ai_tools() -> str:
    """List all tools currently registered in the Wintermute AI tool registry.

    These are the tools available for LLM function-calling via
    tool_calling_chat().

    Returns:
        JSON array of tool definitions (name, description, parameters).
    """
    defs = global_tool_registry.get_definitions()
    return json.dumps(defs, indent=2)

list_target_vulnerabilities(target_id) async

List all vulnerabilities currently registered on a specific target.

The target_id can be any object that holds vulnerabilities: Device, Service, AWSAccount, User, or Peripheral. Use this to know what has already been found before running additional tests.

Parameters:

Name	Type	Description	Default
target_id	str	Registry ID of the target object.	required

Returns:

Type	Description
str	JSON array of vulnerability summaries (title, cvss, vuln_id, threat).

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def list_target_vulnerabilities(target_id: str) -> str:
    """List all vulnerabilities currently registered on a specific target.

    The ``target_id`` can be any object that holds vulnerabilities: Device,
    Service, AWSAccount, User, or Peripheral.  Use this to know what has
    already been found before running additional tests.

    Args:
        target_id: Registry ID of the target object.

    Returns:
        JSON array of vulnerability summaries (title, cvss, vuln_id, threat).
    """
    obj = registry.get(target_id)
    if obj is None:
        return f"ID '{target_id}' not found."
    vulns: list[Vulnerability] = getattr(obj, "vulnerabilities", [])
    summaries = [
        {
            "vuln_id": v.vuln_id,
            "title": v.title,
            "cvss": v.cvss,
            "threat": v.threat,
            "verified": v.verified,
        }
        for v in vulns
    ]
    if not summaries:
        return f"No vulnerabilities on {target_id}."
    return json.dumps(summaries, indent=2)

load_cartridge(cartridge_name) async

Load a cartridge plugin by name.

The cartridge module is imported and its main class (if any) is instantiated. Use list_cartridges() first to see available names.

Parameters:

Name	Type	Description	Default
cartridge_name	str	Module name (e.g. "tpm20").	required

Returns:

Type	Description
str	Confirmation with the cartridge's available methods/classes.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def load_cartridge(cartridge_name: str) -> str:
    """Load a cartridge plugin by name.

    The cartridge module is imported and its main class (if any) is
    instantiated.  Use list_cartridges() first to see available names.

    Args:
        cartridge_name: Module name (e.g. ``"tpm20"``).

    Returns:
        Confirmation with the cartridge's available methods/classes.
    """
    if cartridge_name in _loaded_cartridges:
        return f"Cartridge '{cartridge_name}' is already loaded."
    try:
        mod = importlib.import_module(f"wintermute.cartridges.{cartridge_name}")
        _loaded_cartridges[cartridge_name] = mod
        members = [n for n in dir(mod) if not n.startswith("_")]
        return f"Loaded '{cartridge_name}'. Exports: {members}"
    except Exception as e:
        return f"Failed to load cartridge '{cartridge_name}': {e}"

load_operation(operation_name) async

Load a previously saved Operation from the storage backend.

Creates a new Operation shell, hydrates it from storage, and registers it in the session registry. Also registers all child objects (devices, users, cloud accounts) so they are immediately addressable.

Parameters:

Name	Type	Description	Default
operation_name	str	The operation_name key used when the Operation was saved (NOT a registry ID — this is the raw name).	required

Returns:

Type	Description
str	The registry string ID for the loaded Operation, or an error.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def load_operation(operation_name: str) -> str:
    """Load a previously saved Operation from the storage backend.

    Creates a new Operation shell, hydrates it from storage, and registers it
    in the session registry.  Also registers all child objects (devices, users,
    cloud accounts) so they are immediately addressable.

    Args:
        operation_name: The ``operation_name`` key used when the Operation was
                        saved (NOT a registry ID — this is the raw name).

    Returns:
        The registry string ID for the loaded Operation, or an error.
    """
    try:
        op = Operation(operation_name=operation_name)
        op.load()
    except Exception as e:
        return f"Load failed: {e}"
    oid = registry.store(op, "operation", operation_name, prefix="op")
    # Auto-register children
    for dev in op.devices:
        registry.store(dev, "device", dev.hostname, prefix="dev")
    for user in op.users:
        registry.store(user, "user", user.uid, prefix="user")
    for acc in op.cloud_accounts:
        hint = getattr(acc, "account_id", None) or getattr(acc, "name", "cloud")
        pref = "aws" if isinstance(acc, AWSAccount) else "cloud"
        registry.store(acc, type(acc).__name__, str(hint), prefix=pref)
    return f"Loaded and registered as {oid} with {len(op.devices)} devices, {len(op.users)} users, {len(op.cloud_accounts)} cloud accounts."

main()

CLI entry point for the Wintermute MCP Server.

Source code in wintermute/WintermuteMCP.py

def main() -> None:
    """CLI entry point for the Wintermute MCP Server."""
    parser = argparse.ArgumentParser(
        description="Wintermute MCP Server — Hardware Security AI Agent Interface",
    )
    parser.add_argument(
        "--host",
        default="127.0.0.1",
        help="Bind address (default: 127.0.0.1)",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=31337,
        help="Bind port (default: 31337)",
    )
    parser.add_argument(
        "--transport",
        choices=["sse", "stdio"],
        default="sse",
        help="MCP transport (default: sse)",
    )
    parser.add_argument(
        "--log-level",
        default="INFO",
        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
        help="Logging level (default: INFO)",
    )
    args = parser.parse_args()

    logging.basicConfig(
        level=getattr(logging, args.log_level),
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )

    if args.transport == "sse":
        # Configure FastMCP settings for SSE transport
        mcp.settings.host = args.host
        mcp.settings.port = args.port
        log.info("Starting WintermuteMCP on %s:%d (SSE)", args.host, args.port)
        mcp.run(transport="sse")
    else:
        log.info("Starting WintermuteMCP on stdio")
        mcp.run(transport="stdio")

open_ssh_session(target_alias, username=None, password=None, port=None) async

Open a persistent SSH session for multi-command workflows.

Use this instead of run_ssh_command when you need to run many commands on the same target, launch background fuzzers or long-running tools, or preserve shell state across calls. The session stays open until you explicitly call close_ssh_session.

Parameters:

Name	Type	Description	Default
target_alias	str	SSH host alias from ~/.ssh/config, or hostname/IP.	required
username	str | None	SSH username (optional — inferred from ssh config).	None
password	str | None	SSH password (optional — falls back to key-based auth).	None
port	int | None	SSH port (optional — defaults to ssh config or 22).	None

Returns:

Type	Description
str	A session_id string for use with the other *_ssh_session* tools.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def open_ssh_session(
    target_alias: str,
    username: str | None = None,
    password: str | None = None,
    port: int | None = None,
) -> str:
    """Open a persistent SSH session for multi-command workflows.

    Use this instead of ``run_ssh_command`` when you need to run many commands
    on the same target, launch background fuzzers or long-running tools, or
    preserve shell state across calls.  The session stays open until you
    explicitly call ``close_ssh_session``.

    Args:
        target_alias: SSH host alias from ~/.ssh/config, or hostname/IP.
        username: SSH username (optional — inferred from ssh config).
        password: SSH password (optional — falls back to key-based auth).
        port: SSH port (optional — defaults to ssh config or 22).

    Returns:
        A session_id string for use with the other ``*_ssh_session*`` tools.
    """
    from wintermute.ai.utils.ssh_exec import SSHSession

    try:
        session = SSHSession(
            target_alias=target_alias,
            username=username,
            password=password,
            port=port,
        )
        await session.connect()
        session_id = registry.store(session, "ssh_session", target_alias, prefix="ssh")
        return json.dumps({"session_id": session_id}, indent=2)
    except Exception as e:
        return json.dumps({"error": f"Failed to open SSH session: {e}"}, indent=2)

poll_ssh_background_job(session_id, job_id) async

Poll the status of a background job on a persistent SSH session.

Do NOT block in a loop — call this tool, read the status, and if still "running" wait before calling again. This is the correct pattern for long-running tools.

Parameters:

Name	Type	Description	Default
session_id	str	ID returned by open_ssh_session.	required
job_id	str	ID returned by run_ssh_session_background.	required

Returns:

Type	Description
str	JSON with status ("running", "done", or "error"),
str	and exit_code, stdout, stderr when finished.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def poll_ssh_background_job(session_id: str, job_id: str) -> str:
    """Poll the status of a background job on a persistent SSH session.

    Do NOT block in a loop — call this tool, read the status, and if still
    ``"running"`` wait before calling again.  This is the correct pattern for
    long-running tools.

    Args:
        session_id: ID returned by ``open_ssh_session``.
        job_id: ID returned by ``run_ssh_session_background``.

    Returns:
        JSON with ``status`` (``"running"``, ``"done"``, or ``"error"``),
        and ``exit_code``, ``stdout``, ``stderr`` when finished.
    """
    from wintermute.ai.utils.ssh_exec import SSHSession

    try:
        session: SSHSession = _require(session_id, SSHSession, "SSHSession")
        result = await session.poll_job(job_id)
        return json.dumps(result, indent=2, default=str)
    except (ValueError, TypeError) as e:
        return json.dumps({"error": str(e)}, indent=2)

read_ticket(ticket_id) async

Read the current state of a ticket from the ticketing backend.

Parameters:

Name	Type	Description	Default
ticket_id	str	The ticket ID returned by create_ticket().	required

Returns:

Type	Description
str	JSON representation of the ticket data and comments.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def read_ticket(ticket_id: str) -> str:
    """Read the current state of a ticket from the ticketing backend.

    Args:
        ticket_id: The ticket ID returned by create_ticket().

    Returns:
        JSON representation of the ticket data and comments.
    """
    try:
        t = Ticket.read(ticket_id)
        return _ser(t)
    except Exception as e:
        return f"Failed: {e}"

register_python_tools(function_names_json) async

Register Python functions as AI-callable tools in the global registry.

Each function must be importable and have full type annotations.

Parameters:

Name	Type	Description	Default
function_names_json	str	JSON array of fully-qualified function names (e.g. '["mymodule.my_func"]').	required

Returns:

Type	Description
str	List of registered tool names, or errors.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def register_python_tools(function_names_json: str) -> str:
    """Register Python functions as AI-callable tools in the global registry.

    Each function must be importable and have full type annotations.

    Args:
        function_names_json: JSON array of fully-qualified function names
                             (e.g. ``'["mymodule.my_func"]'``).

    Returns:
        List of registered tool names, or errors.
    """
    from wintermute.ai.utils.tool_factory import register_tools

    try:
        names: list[str] = json.loads(function_names_json)
    except json.JSONDecodeError:
        return "Invalid JSON array."
    functions: list[Any] = []
    for name in names:
        parts = name.rsplit(".", 1)
        if len(parts) != 2:
            return f"Invalid function path: '{name}'. Use 'module.function' format."
        mod = importlib.import_module(parts[0])
        functions.append(getattr(mod, parts[1]))
    tools_created = register_tools(functions)
    for t in tools_created:
        global_tool_registry.register(t)
    return f"Registered {len(tools_created)} tools: {[t.name for t in tools_created]}"

remove_vulnerability_from_target(target_id, vuln_title=None, vuln_id=None) async

Remove a vulnerability from any registered target object.

Provide either vuln_title or vuln_id to identify the vulnerability.

Parameters:

Name	Type	Description	Default
target_id	str	Registry ID of the object holding the vulnerability.	required
vuln_title	str | None	Title of the vulnerability to remove (optional).	None
vuln_id	str | None	UUID of the vulnerability to remove (optional).	None

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def remove_vulnerability_from_target(
    target_id: str,
    vuln_title: str | None = None,
    vuln_id: str | None = None,
) -> str:
    """Remove a vulnerability from any registered target object.

    Provide either ``vuln_title`` or ``vuln_id`` to identify the vulnerability.

    Args:
        target_id: Registry ID of the object holding the vulnerability.
        vuln_title: Title of the vulnerability to remove (optional).
        vuln_id: UUID of the vulnerability to remove (optional).

    Returns:
        Confirmation or error message.
    """
    obj = registry.get(target_id)
    if obj is None:
        return f"ID '{target_id}' not found."
    kwargs: dict[str, str] = {}
    if vuln_id:
        kwargs["uid"] = vuln_id
    elif vuln_title:
        kwargs["title"] = vuln_title
    else:
        return "Provide at least 'vuln_title' or 'vuln_id'."
    result = remove_vulnerability(obj, **kwargs)
    return f"Vulnerability removed: {result}"

run_ssh_command(target_alias, command, username=None, password=None, port=None) async

Execute a command on a remote host via SSH.

Uses asyncssh and the host machine's ~/.ssh/config for proxy jumping, key management, and host aliases.

Parameters:

Name	Type	Description	Default
target_alias	str	SSH host alias from ~/.ssh/config, or hostname/IP.	required
command	str	Shell command to execute on the remote host.	required
username	str | None	SSH username (optional — inferred from ssh config).	None
password	str | None	SSH password (optional — falls back to key-based auth).	None
port	int | None	SSH port (optional — defaults to ssh config or 22).	None

Returns:

Type	Description
str	JSON with stdout, stderr, and exit_code.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def run_ssh_command(
    target_alias: str,
    command: str,
    username: str | None = None,
    password: str | None = None,
    port: int | None = None,
) -> str:
    """Execute a command on a remote host via SSH.

    Uses asyncssh and the host machine's ``~/.ssh/config`` for proxy jumping,
    key management, and host aliases.

    Args:
        target_alias: SSH host alias from ~/.ssh/config, or hostname/IP.
        command: Shell command to execute on the remote host.
        username: SSH username (optional — inferred from ssh config).
        password: SSH password (optional — falls back to key-based auth).
        port: SSH port (optional — defaults to ssh config or 22).

    Returns:
        JSON with ``stdout``, ``stderr``, and ``exit_code``.
    """
    from wintermute.ai.utils.ssh_exec import run_command_async

    result = await run_command_async(
        target_alias=target_alias,
        command=command,
        username=username,
        password=password,
        port=port,
    )
    return json.dumps(result, indent=2, default=str)

run_ssh_session_background(session_id, command) async

Launch a background command on a persistent SSH session.

Use this for fuzzers, long scanners, or any tool that may run for minutes or hours. Poll with poll_ssh_background_job to check completion.

Parameters:

Name	Type	Description	Default
session_id	str	ID returned by open_ssh_session.	required
command	str	Shell command to launch in the background via nohup.	required

Returns:

Type	Description
str	JSON with the job_id to use when polling.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def run_ssh_session_background(session_id: str, command: str) -> str:
    """Launch a background command on a persistent SSH session.

    Use this for fuzzers, long scanners, or any tool that may run for minutes
    or hours.  Poll with ``poll_ssh_background_job`` to check completion.

    Args:
        session_id: ID returned by ``open_ssh_session``.
        command: Shell command to launch in the background via nohup.

    Returns:
        JSON with the ``job_id`` to use when polling.
    """
    from wintermute.ai.utils.ssh_exec import SSHSession

    try:
        session: SSHSession = _require(session_id, SSHSession, "SSHSession")
        job_id = await session.run_background(command)
        return json.dumps({"job_id": job_id}, indent=2)
    except (ValueError, TypeError) as e:
        return json.dumps({"error": str(e)}, indent=2)

run_ssh_session_command(session_id, command) async

Execute a command on a persistent SSH session.

Parameters:

Name	Type	Description	Default
session_id	str	ID returned by open_ssh_session.	required
command	str	Shell command to execute.	required

Returns:

Type	Description
str	JSON with exit_code, stdout, and stderr.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def run_ssh_session_command(session_id: str, command: str) -> str:
    """Execute a command on a persistent SSH session.

    Args:
        session_id: ID returned by ``open_ssh_session``.
        command: Shell command to execute.

    Returns:
        JSON with ``exit_code``, ``stdout``, and ``stderr``.
    """
    from wintermute.ai.utils.ssh_exec import SSHSession

    try:
        session: SSHSession = _require(session_id, SSHSession, "SSHSession")
        result = await session.run(command)
        return json.dumps(result, indent=2, default=str)
    except (ValueError, TypeError) as e:
        return json.dumps({"error": str(e)}, indent=2)

save_operation(operation_id) async

Persist an Operation to the currently registered storage backend.

You MUST call setup_storage_backend() first. The Operation is saved using its operation_name as the storage key.

Parameters:

Name	Type	Description	Default
operation_id	str	The registry string ID of the Operation to save.	required

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def save_operation(operation_id: str) -> str:
    """Persist an Operation to the currently registered storage backend.

    You MUST call setup_storage_backend() first.  The Operation is saved using
    its ``operation_name`` as the storage key.

    Args:
        operation_id: The registry string ID of the Operation to save.

    Returns:
        Confirmation or error message.
    """
    op: Operation = _require(operation_id, Operation, "Operation")
    try:
        result = op.save()
        return f"Saved '{op.operation_name}': {result}"
    except Exception as e:
        return f"Save failed: {e}"

set_ai_default_provider(provider=None, model=None) async

Switch the AI router to a different LLM provider or model.

You MUST have called configure_ai_router() first. Use this to toggle between base providers ("bedrock") and RAG-augmented providers ("rag-tiny_hardware_test"). Setting provider to a RAG name makes all subsequent AI queries context-aware.

Parameters:

Name	Type	Description	Default
provider	str | None	Provider name from the registry (e.g. "bedrock", "groq", "rag-my_kb"). Empty string keeps current.	None
model	str | None	Model ID override. Empty string keeps current.	None

Returns:

Type	Description
str	Confirmation of the new default settings.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def set_ai_default_provider(
    provider: str | None = None,
    model: str | None = None,
) -> str:
    """Switch the AI router to a different LLM provider or model.

    You MUST have called configure_ai_router() first.  Use this to toggle
    between base providers (``"bedrock"``) and RAG-augmented providers
    (``"rag-tiny_hardware_test"``).  Setting ``provider`` to a RAG name makes
    all subsequent AI queries context-aware.

    Args:
        provider: Provider name from the registry (e.g. ``"bedrock"``,
                  ``"groq"``, ``"rag-my_kb"``).  Empty string keeps current.
        model: Model ID override.  Empty string keeps current.

    Returns:
        Confirmation of the new default settings.
    """
    router = _router()
    if router is None:
        return "AI router not initialized. Call configure_ai_router() first."
    if not provider and not model:
        return "Provide at least one of 'provider' or 'model'."
    router.set_default(
        provider=provider if provider else None,
        model=model if model else None,
    )
    return f"AI default updated: provider={router.default_provider}, model={router.default_model}"

set_device_processor(device_id, processor_name, manufacturer=None, instruction_set=None, core=None, cpu_cores=None, endianness=None) async

Assign a Processor and Architecture to a Device.

Use this after identifying the target's CPU (e.g. from board markings, firmware headers, or JTAG IDCODE).

Parameters:

Name	Type	Description	Default
device_id	str	Registry ID of the target Device.	required
processor_name	str	Processor name/model (e.g. "Cortex-A7").	required
manufacturer	str | None	Manufacturer (e.g. "ARM").	None
instruction_set	str | None	ISA (e.g. "ARMv7-A").	None
core	str | None	Core type (e.g. "Cortex-A7").	None
cpu_cores	int | None	Number of CPU cores.	None
endianness	str | None	"little" or "big".	None

Returns:

Type	Description
str	Confirmation message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def set_device_processor(
    device_id: str,
    processor_name: str,
    manufacturer: str | None = None,
    instruction_set: str | None = None,
    core: str | None = None,
    cpu_cores: int | None = None,
    endianness: str | None = None,
) -> str:
    """Assign a Processor and Architecture to a Device.

    Use this after identifying the target's CPU (e.g. from board markings,
    firmware headers, or JTAG IDCODE).

    Args:
        device_id: Registry ID of the target Device.
        processor_name: Processor name/model (e.g. ``"Cortex-A7"``).
        manufacturer: Manufacturer (e.g. ``"ARM"``).
        instruction_set: ISA (e.g. ``"ARMv7-A"``).
        core: Core type (e.g. ``"Cortex-A7"``).
        cpu_cores: Number of CPU cores.
        endianness: ``"little"`` or ``"big"``.

    Returns:
        Confirmation message.
    """
    from wintermute.core import Device

    dev: Device = _require(device_id, Device, "Device")
    arch_kwargs: dict[str, Any] = {}
    if core:
        arch_kwargs["core"] = core
    if instruction_set:
        arch_kwargs["instruction_set"] = instruction_set
    if cpu_cores:
        arch_kwargs["cpu_cores"] = cpu_cores
    arch = Architecture(**arch_kwargs) if arch_kwargs else None
    proc = Processor(
        processor=processor_name,
        manufacturer=manufacturer or None,
        architecture=arch or {},
        endianness=endianness or None,
    )
    dev.processor = proc
    if arch:
        dev.architecture = arch
    return f"Processor '{processor_name}' assigned to {device_id}."

setup_report_backend(template_dir=None, main_template=None, vuln_template=None, test_run_template=None) async

Register a DOCX report backend for automated report generation.

After calling this, use generate_report() to produce Word documents.

Parameters:

Name	Type	Description	Default
template_dir	str | None	Path to the directory containing .docx templates.	None
main_template	str | None	Filename of the main report template.	None
vuln_template	str | None	Filename of the per-vulnerability template.	None
test_run_template	str | None	Filename of the per-test-run template.	None

Returns:

Type	Description
str	Confirmation message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def setup_report_backend(
    template_dir: str | None = None,
    main_template: str | None = None,
    vuln_template: str | None = None,
    test_run_template: str | None = None,
) -> str:
    """Register a DOCX report backend for automated report generation.

    After calling this, use ``generate_report()`` to produce Word documents.

    Args:
        template_dir: Path to the directory containing .docx templates.
        main_template: Filename of the main report template.
        vuln_template: Filename of the per-vulnerability template.
        test_run_template: Filename of the per-test-run template.

    Returns:
        Confirmation message.
    """
    from wintermute.backends.docx_reports import DocxTplPerVulnBackend

    backend = DocxTplPerVulnBackend(
        template_dir=template_dir or "templates",
        main_template=main_template or "report_main.docx",
        vuln_template=vuln_template or "report_vuln.docx",
        test_run_template=test_run_template or "report_test_run.docx",
    )
    Report.register_backend("docx", backend, make_default=True)
    return f"Registered DocxTplPerVulnBackend from '{template_dir}' as default."

setup_storage_backend(backend_type=None, base_path=None, table_name=None, region=None) async

Register a persistence backend for Operation save/load.

Supported backend types: - "json" — Local JSON flat-file storage (default). - "dynamodb" — AWS DynamoDB cloud storage (requires AWS credentials).

After calling this, use save_operation() and load_operation() to persist and restore Operations.

Parameters:

Name	Type	Description	Default
backend_type	str | None	"json" or "dynamodb".	None
base_path	str | None	Directory for JSON files (only used with "json").	None
table_name	str | None	DynamoDB table name (only used with "dynamodb").	None
region	str | None	AWS region (only used with "dynamodb").	None

Returns:

Type	Description
str	Confirmation with the backend name.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def setup_storage_backend(
    backend_type: str | None = None,
    base_path: str | None = None,
    table_name: str | None = None,
    region: str | None = None,
) -> str:
    """Register a persistence backend for Operation save/load.

    Supported backend types:
    - ``"json"`` — Local JSON flat-file storage (default).
    - ``"dynamodb"`` — AWS DynamoDB cloud storage (requires AWS credentials).

    After calling this, use ``save_operation()`` and ``load_operation()`` to
    persist and restore Operations.

    Args:
        backend_type: ``"json"`` or ``"dynamodb"``.
        base_path: Directory for JSON files (only used with ``"json"``).
        table_name: DynamoDB table name (only used with ``"dynamodb"``).
        region: AWS region (only used with ``"dynamodb"``).

    Returns:
        Confirmation with the backend name.
    """
    backend_type = backend_type or "json"
    base_path = base_path or ".wintermute_data"
    table_name = table_name or "WintermuteOperations"
    region = region or "us-east-1"
    if backend_type == "json":
        backend = JsonFileBackend(base_path=base_path)
        Operation.register_backend("json", backend, make_default=True)
        return f"Registered JsonFileBackend at '{base_path}' as default."
    elif backend_type == "dynamodb":
        from wintermute.backends.dynamodb import DynamoDBBackend

        backend_ddb = DynamoDBBackend(
            table_name=table_name, region_name=region, create_if_missing=True
        )
        Operation.register_backend("dynamodb", backend_ddb, make_default=True)
        return f"Registered DynamoDBBackend (table={table_name}, region={region}) as default."
    return f"Unknown backend type: '{backend_type}'. Use 'json' or 'dynamodb'."

setup_ticket_backend(backend_type=None, base_url=None, api_key=None, default_product=None, default_component=None) async

Register a ticketing backend for vulnerability and incident tracking.

Supported backend types: - "memory" — In-memory store (useful for testing / ephemeral sessions). - "bugzilla" — Bugzilla REST API (requires base_url and api_key).

After calling this, use create_ticket(), read_ticket(), update_ticket() to manage tickets.

Parameters:

Name	Type	Description	Default
backend_type	str | None	"memory" or "bugzilla".	None
base_url	str | None	Bugzilla REST endpoint (e.g. "http://host/bugzilla/rest").	None
api_key	str | None	Bugzilla API key.	None
default_product	str | None	Default Bugzilla product for new tickets.	None
default_component	str | None	Default Bugzilla component for new tickets.	None

Returns:

Type	Description
str	Confirmation with the backend name.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def setup_ticket_backend(
    backend_type: str | None = None,
    base_url: str | None = None,
    api_key: str | None = None,
    default_product: str | None = None,
    default_component: str | None = None,
) -> str:
    """Register a ticketing backend for vulnerability and incident tracking.

    Supported backend types:
    - ``"memory"`` — In-memory store (useful for testing / ephemeral sessions).
    - ``"bugzilla"`` — Bugzilla REST API (requires ``base_url`` and ``api_key``).

    After calling this, use ``create_ticket()``, ``read_ticket()``,
    ``update_ticket()`` to manage tickets.

    Args:
        backend_type: ``"memory"`` or ``"bugzilla"``.
        base_url: Bugzilla REST endpoint (e.g. ``"http://host/bugzilla/rest"``).
        api_key: Bugzilla API key.
        default_product: Default Bugzilla product for new tickets.
        default_component: Default Bugzilla component for new tickets.

    Returns:
        Confirmation with the backend name.
    """
    backend_type = backend_type or "memory"
    if backend_type == "memory":
        Ticket.register_backend("memory", InMemoryBackend(), make_default=True)
        return "Registered InMemoryBackend as default ticket backend."
    elif backend_type == "bugzilla":
        if not base_url or not api_key:
            return "Bugzilla backend requires 'base_url' and 'api_key'."
        from wintermute.backends.bugzilla import BugzillaBackend

        bz = BugzillaBackend(
            base_url=base_url,
            api_key=api_key,
            default_product=default_product or None,
            default_component=default_component or None,
        )
        Ticket.register_backend("bugzilla", bz, make_default=True)
        return f"Registered BugzillaBackend at '{base_url}' as default."
    return f"Unknown backend type: '{backend_type}'. Use 'memory' or 'bugzilla'."

unload_cartridge(cartridge_name) async

Unload a previously loaded cartridge.

Parameters:

Name	Type	Description	Default
cartridge_name	str	Module name to unload.	required

Returns:

Type	Description
str	Confirmation or error.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def unload_cartridge(cartridge_name: str) -> str:
    """Unload a previously loaded cartridge.

    Args:
        cartridge_name: Module name to unload.

    Returns:
        Confirmation or error.
    """
    if cartridge_name in _loaded_cartridges:
        del _loaded_cartridges[cartridge_name]
        fqn = f"wintermute.cartridges.{cartridge_name}"
        sys.modules.pop(fqn, None)
        return f"Unloaded '{cartridge_name}'."
    return f"Cartridge '{cartridge_name}' is not loaded."

update_test_run_status(test_run_id, status, executed_by=None, notes=None) async

Update the execution status of a test case run.

Valid statuses: not_run, in_progress, passed, failed, blocked, not_applicable.

Setting status to in_progress also calls start() (sets started_at). Any terminal status calls finish() (sets ended_at).

Parameters:

Name	Type	Description	Default
test_run_id	str	Registry ID of the TestCaseRun.	required
status	str	New status string (see above).	required
executed_by	str | None	Analyst userid who performed the test.	None
notes	str | None	Free-text execution notes.	None

Returns:

Type	Description
str	Confirmation with the new status.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def update_test_run_status(
    test_run_id: str,
    status: str,
    executed_by: str | None = None,
    notes: str | None = None,
) -> str:
    """Update the execution status of a test case run.

    Valid statuses: ``not_run``, ``in_progress``, ``passed``, ``failed``,
    ``blocked``, ``not_applicable``.

    Setting status to ``in_progress`` also calls ``start()`` (sets
    ``started_at``).  Any terminal status calls ``finish()`` (sets
    ``ended_at``).

    Args:
        test_run_id: Registry ID of the TestCaseRun.
        status: New status string (see above).
        executed_by: Analyst userid who performed the test.
        notes: Free-text execution notes.

    Returns:
        Confirmation with the new status.
    """
    run: TestCaseRun = _require(test_run_id, TestCaseRun, "TestCaseRun")
    try:
        new_status = RunStatus(status)
    except ValueError:
        return f"Invalid status '{status}'. Valid: {[s.value for s in RunStatus]}"
    run.status = new_status
    if new_status == RunStatus.in_progress:
        run.start()
    elif new_status in (
        RunStatus.passed,
        RunStatus.failed,
        RunStatus.blocked,
        RunStatus.not_applicable,
    ):
        run.finish()
    if executed_by:
        run.executed_by = executed_by
    if notes:
        run.notes = notes
    return f"Run {test_run_id} -> {new_status.value}"

update_ticket(ticket_id, status=None, assignee=None, description=None) async

Update fields on an existing ticket.

Parameters:

Name	Type	Description	Default
ticket_id	str	The ticket ID to update.	required
status	str | None	New status — one of "open", "in_progress", "resolved", "closed" (empty to skip).	None
assignee	str | None	New assignee (empty to skip).	None
description	str | None	Append a note / new description (empty to skip).	None

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def update_ticket(
    ticket_id: str,
    status: str | None = None,
    assignee: str | None = None,
    description: str | None = None,
) -> str:
    """Update fields on an existing ticket.

    Args:
        ticket_id: The ticket ID to update.
        status: New status — one of ``"open"``, ``"in_progress"``,
                ``"resolved"``, ``"closed"`` (empty to skip).
        assignee: New assignee (empty to skip).
        description: Append a note / new description (empty to skip).

    Returns:
        Confirmation or error message.
    """
    try:
        fields: dict[str, Any] = {}
        if status:
            fields["status"] = Status(status)
        if assignee:
            fields["assignee"] = assignee
        if description:
            fields["description"] = description
        Ticket.update(ticket_id, **fields)
        return f"Updated ticket {ticket_id}."
    except Exception as e:
        return f"Failed: {e}"

upload_file_ssh(target_alias, local_path, remote_path, username=None, password=None, port=None) async

Upload a file to a remote host via SFTP/SSH.

Uses asyncssh and the host machine's ~/.ssh/config for proxy jumping, key management, and host aliases.

Parameters:

Name	Type	Description	Default
target_alias	str	SSH host alias from ~/.ssh/config, or hostname/IP.	required
local_path	str	Local file path to upload.	required
remote_path	str	Destination path on the remote host.	required
username	str | None	SSH username (optional — inferred from ssh config).	None
password	str | None	SSH password (optional — falls back to key-based auth).	None
port	int | None	SSH port (optional — defaults to ssh config or 22).	None

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def upload_file_ssh(
    target_alias: str,
    local_path: str,
    remote_path: str,
    username: str | None = None,
    password: str | None = None,
    port: int | None = None,
) -> str:
    """Upload a file to a remote host via SFTP/SSH.

    Uses asyncssh and the host machine's ``~/.ssh/config`` for proxy jumping,
    key management, and host aliases.

    Args:
        target_alias: SSH host alias from ~/.ssh/config, or hostname/IP.
        local_path: Local file path to upload.
        remote_path: Destination path on the remote host.
        username: SSH username (optional — inferred from ssh config).
        password: SSH password (optional — falls back to key-based auth).
        port: SSH port (optional — defaults to ssh config or 22).

    Returns:
        Confirmation or error message.
    """
    from wintermute.ai.utils.ssh_exec import upload_file_async

    result = await upload_file_async(
        target_alias=target_alias,
        local_path=local_path,
        remote_path=remote_path,
        username=username,
        password=password,
        port=port,
    )
    return json.dumps(result, indent=2, default=str)

upload_file_ssh_session(session_id, local_path, remote_path) async

Upload a file via SFTP on a persistent SSH session.

Parameters:

Name	Type	Description	Default
session_id	str	ID returned by open_ssh_session.	required
local_path	str	Local file path to upload.	required
remote_path	str	Destination path on the remote host.	required

Returns:

Type	Description
str	Confirmation or error message.

Source code in wintermute/WintermuteMCP.py

@mcp.tool()
async def upload_file_ssh_session(
    session_id: str, local_path: str, remote_path: str
) -> str:
    """Upload a file via SFTP on a persistent SSH session.

    Args:
        session_id: ID returned by ``open_ssh_session``.
        local_path: Local file path to upload.
        remote_path: Destination path on the remote host.

    Returns:
        Confirmation or error message.
    """
    from wintermute.ai.utils.ssh_exec import SSHSession

    try:
        session: SSHSession = _require(session_id, SSHSession, "SSHSession")
        result = await session.upload(local_path, remote_path)
        return json.dumps(result, indent=2, default=str)
    except (ValueError, TypeError) as e:
        return json.dumps({"error": str(e)}, indent=2)