McpPlugin

McpPlugin

McpPlugin(verifier)

Bases: BasePlugin

MCP interception plugin.

Patches ClientSession.call_tool, ClientSession.read_resource, ClientSession.get_prompt and Server._handle_request at the class level. Uses reference counting so nested sandboxes work correctly.

Each (direction, method, key) triple has its own FIFO deque of McpMockConfig objects.

Source code in src/tripwire/plugins/mcp_plugin.py

def __init__(self, verifier: StrictVerifier) -> None:
    super().__init__(verifier)
    self._queues: dict[str, deque[McpMockConfig]] = {}
    self._registry_lock: threading.Lock = threading.Lock()

mock_call_tool

mock_call_tool(tool_name, *, returns, raises=None, required=True)

Register a mock for a client call_tool invocation.

Source code in src/tripwire/plugins/mcp_plugin.py

def mock_call_tool(
    self,
    tool_name: str,
    *,
    returns: Any,  # noqa: ANN401
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a mock for a client call_tool invocation."""
    self._enqueue(
        "client", "call_tool", tool_name,
        returns=returns, raises=raises, required=required,
    )

mock_read_resource

mock_read_resource(uri, *, returns, raises=None, required=True)

Register a mock for a client read_resource invocation.

Source code in src/tripwire/plugins/mcp_plugin.py

def mock_read_resource(
    self,
    uri: str,
    *,
    returns: Any,  # noqa: ANN401
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a mock for a client read_resource invocation."""
    self._enqueue(
        "client", "read_resource", uri,
        returns=returns, raises=raises, required=required,
    )

mock_get_prompt

mock_get_prompt(prompt_name, *, returns, raises=None, required=True)

Register a mock for a client get_prompt invocation.

Source code in src/tripwire/plugins/mcp_plugin.py

def mock_get_prompt(
    self,
    prompt_name: str,
    *,
    returns: Any,  # noqa: ANN401
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a mock for a client get_prompt invocation."""
    self._enqueue(
        "client", "get_prompt", prompt_name,
        returns=returns, raises=raises, required=required,
    )

mock_server_call_tool

mock_server_call_tool(tool_name, *, returns, raises=None, required=True)

Register a mock for a server call_tool handler invocation.

Source code in src/tripwire/plugins/mcp_plugin.py

def mock_server_call_tool(
    self,
    tool_name: str,
    *,
    returns: Any,  # noqa: ANN401
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a mock for a server call_tool handler invocation."""
    self._enqueue(
        "server", "call_tool", tool_name,
        returns=returns, raises=raises, required=required,
    )

mock_server_read_resource

mock_server_read_resource(uri, *, returns, raises=None, required=True)

Register a mock for a server read_resource handler invocation.

Source code in src/tripwire/plugins/mcp_plugin.py

def mock_server_read_resource(
    self,
    uri: str,
    *,
    returns: Any,  # noqa: ANN401
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a mock for a server read_resource handler invocation."""
    self._enqueue(
        "server", "read_resource", uri,
        returns=returns, raises=raises, required=required,
    )

mock_server_get_prompt

mock_server_get_prompt(prompt_name, *, returns, raises=None, required=True)

Register a mock for a server get_prompt handler invocation.

Source code in src/tripwire/plugins/mcp_plugin.py

def mock_server_get_prompt(
    self,
    prompt_name: str,
    *,
    returns: Any,  # noqa: ANN401
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a mock for a server get_prompt handler invocation."""
    self._enqueue(
        "server", "get_prompt", prompt_name,
        returns=returns, raises=raises, required=required,
    )

install_patches

install_patches()

Install MCP client/server patches.

Source code in src/tripwire/plugins/mcp_plugin.py

def install_patches(self) -> None:
    """Install MCP client/server patches."""
    if not _MCP_AVAILABLE:
        raise ImportError(
            "Install python-tripwire[mcp] to use McpPlugin: pip install python-tripwire[mcp]"
        )
    McpPlugin._original_call_tool = _ClientSession.call_tool
    McpPlugin._original_read_resource = _ClientSession.read_resource
    McpPlugin._original_get_prompt = _ClientSession.get_prompt
    McpPlugin._original_handle_request = _Server._handle_request

    setattr(_ClientSession, "call_tool", _patched_call_tool)
    setattr(_ClientSession, "read_resource", _patched_read_resource)
    setattr(_ClientSession, "get_prompt", _patched_get_prompt)
    setattr(_Server, "_handle_request", _patched_handle_request)

restore_patches

restore_patches()

Restore original MCP client/server functions.

Source code in src/tripwire/plugins/mcp_plugin.py

def restore_patches(self) -> None:
    """Restore original MCP client/server functions."""
    if McpPlugin._original_call_tool is not None:
        setattr(_ClientSession, "call_tool", McpPlugin._original_call_tool)
        McpPlugin._original_call_tool = None
    if McpPlugin._original_read_resource is not None:
        setattr(_ClientSession, "read_resource", McpPlugin._original_read_resource)
        McpPlugin._original_read_resource = None
    if McpPlugin._original_get_prompt is not None:
        setattr(_ClientSession, "get_prompt", McpPlugin._original_get_prompt)
        McpPlugin._original_get_prompt = None
    if McpPlugin._original_handle_request is not None:
        setattr(_Server, "_handle_request", McpPlugin._original_handle_request)
        McpPlugin._original_handle_request = None

matches

matches(interaction, expected)

Field-by-field comparison with dirty-equals support.

Source code in src/tripwire/plugins/mcp_plugin.py

def matches(self, interaction: Interaction, expected: dict[str, Any]) -> bool:
    """Field-by-field comparison with dirty-equals support."""
    try:
        for key, expected_val in expected.items():
            actual_val = interaction.details.get(key)
            if expected_val != actual_val:
                return False
        return True
    except Exception:
        return False

get_unused_mocks

get_unused_mocks()

Return all McpMockConfig with required=True still in any queue.

Source code in src/tripwire/plugins/mcp_plugin.py

def get_unused_mocks(self) -> list[McpMockConfig]:
    """Return all McpMockConfig with required=True still in any queue."""
    unused: list[McpMockConfig] = []
    with self._registry_lock:
        for queue in self._queues.values():
            for config in queue:
                if config.required:
                    unused.append(config)
    return unused

assert_call_tool

assert_call_tool(tool_name, *, arguments=None, direction='client', raised=_ABSENT)

Assert the next call_tool interaction.

Source code in src/tripwire/plugins/mcp_plugin.py

def assert_call_tool(
    self,
    tool_name: str,
    *,
    arguments: dict[str, Any] | None = None,
    direction: str = "client",
    raised: Any = _ABSENT,  # noqa: ANN401
) -> None:
    """Assert the next call_tool interaction."""
    from tripwire._context import _get_test_verifier_or_raise  # noqa: PLC0415

    source_id = f"mcp:{direction}:call_tool:{tool_name}"
    sentinel = _McpSentinel(source_id)
    expected: dict[str, Any] = {
        "direction": direction,
        "method": "call_tool",
        "tool_name": tool_name,
        "arguments": arguments if arguments is not None else {},
    }
    if raised is not McpPlugin._ABSENT:
        expected["raised"] = raised
    _get_test_verifier_or_raise().assert_interaction(sentinel, **expected)

assert_read_resource

assert_read_resource(uri, *, direction='client', raised=_ABSENT)

Assert the next read_resource interaction.

Source code in src/tripwire/plugins/mcp_plugin.py

def assert_read_resource(
    self,
    uri: str,
    *,
    direction: str = "client",
    raised: Any = _ABSENT,  # noqa: ANN401
) -> None:
    """Assert the next read_resource interaction."""
    from tripwire._context import _get_test_verifier_or_raise  # noqa: PLC0415

    source_id = f"mcp:{direction}:read_resource:{uri}"
    sentinel = _McpSentinel(source_id)
    expected: dict[str, Any] = {
        "direction": direction,
        "method": "read_resource",
        "uri": uri,
    }
    if raised is not McpPlugin._ABSENT:
        expected["raised"] = raised
    _get_test_verifier_or_raise().assert_interaction(sentinel, **expected)

assert_get_prompt

assert_get_prompt(prompt_name, *, arguments=None, direction='client', raised=_ABSENT)

Assert the next get_prompt interaction.

Source code in src/tripwire/plugins/mcp_plugin.py

def assert_get_prompt(
    self,
    prompt_name: str,
    *,
    arguments: dict[str, Any] | None = None,
    direction: str = "client",
    raised: Any = _ABSENT,  # noqa: ANN401
) -> None:
    """Assert the next get_prompt interaction."""
    from tripwire._context import _get_test_verifier_or_raise  # noqa: PLC0415

    source_id = f"mcp:{direction}:get_prompt:{prompt_name}"
    sentinel = _McpSentinel(source_id)
    expected: dict[str, Any] = {
        "direction": direction,
        "method": "get_prompt",
        "prompt_name": prompt_name,
        "arguments": arguments if arguments is not None else {},
    }
    if raised is not McpPlugin._ABSENT:
        expected["raised"] = raised
    _get_test_verifier_or_raise().assert_interaction(sentinel, **expected)

McpMockConfig

McpMockConfig dataclass

McpMockConfig(direction, method, key, returns, raises=None, required=True, registration_traceback=(lambda: join(format_stack()))())

Configuration for a single mocked MCP call invocation.

Attributes: direction: "client" or "server". method: One of "call_tool", "read_resource", "get_prompt". key: The tool_name, uri, or prompt_name used for queue keying. returns: The value to return when this mock is consumed. raises: If not None, this exception is raised instead of returning. required: If True, the mock is reported as unused if never triggered. registration_traceback: Captured automatically at creation time.