MCP¶

The MCP protocol assertables and content blocks. See the MCP guide for examples and the full method catalog.

pyssertive.protocols.mcp.AssertableMCP ¶

AssertableMCP(payload: Any)

Source code in src/pyssertive/protocols/mcp/assertable.py

def __init__(self, payload: Any) -> None:
    if isinstance(payload, dict):
        self._envelope = _validate_and_wrap(payload)
    elif (
        hasattr(payload, "content")
        and hasattr(payload, "headers")
        and not isinstance(payload, (bytes, bytearray, str))
    ):
        headers = payload.headers
        try:
            ct = headers.get("Content-Type") or headers.get("content-type")
        except AttributeError:
            ct = None
        self._envelope = _parse_text(_coerce_to_text(payload.content), content_type=ct)
    else:
        self._envelope = _parse_text(_coerce_to_text(payload), content_type=None)

negotiated_protocol ¶

negotiated_protocol(expected: str) -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def negotiated_protocol(self, expected: str) -> Self:
    actual = self._require_result().get("protocolVersion")
    if actual != expected:
        raise AssertionError(f"Expected protocolVersion {expected!r}, got {actual!r}")
    return self

server_named ¶

server_named(expected: str) -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def server_named(self, expected: str) -> Self:
    info = self._require_result().get("serverInfo") or {}
    actual = info.get("name")
    if actual != expected:
        raise AssertionError(f"Expected serverInfo.name {expected!r}, got {actual!r}")
    return self

server_version ¶

server_version(expected: str) -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def server_version(self, expected: str) -> Self:
    info = self._require_result().get("serverInfo") or {}
    actual = info.get("version")
    if actual != expected:
        raise AssertionError(f"Expected serverInfo.version {expected!r}, got {actual!r}")
    return self

supports_tools ¶

supports_tools() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def supports_tools(self) -> Self:
    if "tools" not in self._capabilities():
        raise AssertionError("Server does not advertise 'tools' capability")
    return self

supports_resources ¶

supports_resources(
    *, subscribe: bool | None = None
) -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def supports_resources(self, *, subscribe: bool | None = None) -> Self:
    caps = self._capabilities()
    if "resources" not in caps:
        raise AssertionError("Server does not advertise 'resources' capability")
    if subscribe is not None:
        actual = bool((caps.get("resources") or {}).get("subscribe"))
        if actual != subscribe:
            raise AssertionError(f"Expected resources.subscribe={subscribe}, got {actual}")
    return self

supports_prompts ¶

supports_prompts() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def supports_prompts(self) -> Self:
    if "prompts" not in self._capabilities():
        raise AssertionError("Server does not advertise 'prompts' capability")
    return self

supports_logging ¶

supports_logging() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def supports_logging(self) -> Self:
    if "logging" not in self._capabilities():
        raise AssertionError("Server does not advertise 'logging' capability")
    return self

has_instructions ¶

has_instructions() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def has_instructions(self) -> Self:
    if not self._require_result().get("instructions"):
        raise AssertionError("Initialize response has no instructions text")
    return self

lists_tools ¶

lists_tools() -> AssertableToolList

Source code in src/pyssertive/protocols/mcp/assertable.py

def lists_tools(self) -> AssertableToolList:
    return AssertableToolList(self._require_result())

lists_prompts ¶

lists_prompts() -> AssertablePromptList

Source code in src/pyssertive/protocols/mcp/assertable.py

def lists_prompts(self) -> AssertablePromptList:
    return AssertablePromptList(self._require_result())

prompt ¶

prompt(name: str) -> AssertablePromptGet

Source code in src/pyssertive/protocols/mcp/assertable.py

def prompt(self, name: str) -> AssertablePromptGet:
    return AssertablePromptGet(
        name=name,
        result=self._envelope.result if self._envelope.is_success else None,
        error=self._envelope.error if self._envelope.has_error else None,
    )

is_prompts_list_changed_notification ¶

is_prompts_list_changed_notification() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def is_prompts_list_changed_notification(self) -> Self:
    method = self._envelope.raw.get("method")
    if method != "notifications/prompts/list_changed":
        raise AssertionError(
            f"Expected MCP notification 'notifications/prompts/list_changed', got method={method!r}"
        )
    if "id" in self._envelope.raw:
        raise AssertionError(
            f"Expected MCP notification (no 'id' field), got envelope with id={self._envelope.raw['id']!r}"
        )
    return self

tool ¶

tool(name: str) -> AssertableToolCall

Source code in src/pyssertive/protocols/mcp/assertable.py

def tool(self, name: str) -> AssertableToolCall:
    return AssertableToolCall(
        name=name,
        result=self._envelope.result if self._envelope.is_success else None,
        error=self._envelope.error if self._envelope.has_error else None,
    )

is_rejected_with_code ¶

is_rejected_with_code(code: int) -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def is_rejected_with_code(self, code: int) -> Self:
    return self._expect_error_code(code)

is_rejected_as_parse_error ¶

is_rejected_as_parse_error() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def is_rejected_as_parse_error(self) -> Self:
    return self._expect_error_code(ErrorCode.PARSE_ERROR)

is_rejected_as_invalid_request ¶

is_rejected_as_invalid_request() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def is_rejected_as_invalid_request(self) -> Self:
    return self._expect_error_code(ErrorCode.INVALID_REQUEST)

is_rejected_as_method_not_found ¶

is_rejected_as_method_not_found() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def is_rejected_as_method_not_found(self) -> Self:
    return self._expect_error_code(ErrorCode.METHOD_NOT_FOUND)

is_rejected_with_invalid_params ¶

is_rejected_with_invalid_params() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def is_rejected_with_invalid_params(self) -> Self:
    return self._expect_error_code(ErrorCode.INVALID_PARAMS)

is_rejected_as_internal_error ¶

is_rejected_as_internal_error() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def is_rejected_as_internal_error(self) -> Self:
    return self._expect_error_code(ErrorCode.INTERNAL_ERROR)

is_rejected_as_resource_not_found ¶

is_rejected_as_resource_not_found() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def is_rejected_as_resource_not_found(self) -> Self:
    return self._expect_error_code(ErrorCode.RESOURCE_NOT_FOUND)

is_rejected_as_user_rejected ¶

is_rejected_as_user_rejected() -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def is_rejected_as_user_rejected(self) -> Self:
    return self._expect_error_code(ErrorCode.USER_REJECTED)

because_message_contains ¶

because_message_contains(substr: str) -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def because_message_contains(self, substr: str) -> Self:
    if not self._envelope.has_error:
        raise AssertionError("because_message_contains() requires an error envelope")
    message = self._envelope.error_message
    if substr not in message:
        raise AssertionError(f"Error message does not contain {substr!r}: {message!r}")
    return self

because_message_equals ¶

because_message_equals(expected: str) -> Self

Source code in src/pyssertive/protocols/mcp/assertable.py

def because_message_equals(self, expected: str) -> Self:
    if not self._envelope.has_error:
        raise AssertionError("because_message_equals() requires an error envelope")
    message = self._envelope.error_message
    if message != expected:
        raise AssertionError(f"Error message does not match {expected!r}: {message!r}")
    return self

pyssertive.protocols.mcp.AssertableToolList ¶

AssertableToolList(result: dict[str, Any])

Source code in src/pyssertive/protocols/mcp/tools.py

def __init__(self, result: dict[str, Any]) -> None:
    self._result = result
    tools = result.get("tools")
    if not isinstance(tools, list):
        raise AssertionError("Response result is missing 'tools' list — not a tools/list response")
    self._tools: list[dict[str, Any]] = tools

with_count ¶

with_count(expected: int) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def with_count(self, expected: int) -> Self:
    actual = len(self._tools)
    if actual != expected:
        raise AssertionError(f"Expected {expected} tools, got {actual}")
    return self

contains_tool ¶

contains_tool(
    name: str,
    callback: Callable[[AssertableToolDef], Any]
    | None = None,
) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def contains_tool(
    self,
    name: str,
    callback: Callable[[AssertableToolDef], Any] | None = None,
) -> Self:
    match = next((t for t in self._tools if isinstance(t, dict) and t.get("name") == name), None)
    if match is None:
        names = [t.get("name") for t in self._tools if isinstance(t, dict)]
        raise AssertionError(f"Tool '{name}' not found in tools list. Available: {names!r}")
    if callback is not None:
        callback(AssertableToolDef(match))
    return self

does_not_contain_tool ¶

does_not_contain_tool(name: str) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def does_not_contain_tool(self, name: str) -> Self:
    if any(isinstance(t, dict) and t.get("name") == name for t in self._tools):
        raise AssertionError(f"Tool '{name}' should not be in tools list, but it was found")
    return self

every_tool ¶

every_tool(
    callback: Callable[[AssertableToolDef], Any],
) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def every_tool(self, callback: Callable[[AssertableToolDef], Any]) -> Self:
    for tool in self._tools:
        if isinstance(tool, dict):
            callback(AssertableToolDef(tool))
    return self

has_more_pages ¶

has_more_pages() -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def has_more_pages(self) -> Self:
    if not self._result.get("nextCursor"):
        raise AssertionError("Expected tools/list response to advertise a nextCursor")
    return self

pyssertive.protocols.mcp.AssertableToolDef ¶

AssertableToolDef(definition: dict[str, Any])

Source code in src/pyssertive/protocols/mcp/tools.py

def __init__(self, definition: dict[str, Any]) -> None:
    self._definition = definition

documented ¶

documented() -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def documented(self) -> Self:
    description = self._definition.get("description")
    if not description:
        raise AssertionError(f"Tool '{self._name}' has no description")
    return self

accepts ¶

accepts(params: list[str]) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def accepts(self, params: list[str]) -> Self:
    required = list((self._definition.get("inputSchema") or {}).get("required") or [])
    missing = [p for p in params if p not in required]
    if missing:
        raise AssertionError(f"Tool '{self._name}' does not require parameters {missing!r}; required={required!r}")
    return self

accepts_optional ¶

accepts_optional(params: list[str]) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def accepts_optional(self, params: list[str]) -> Self:
    properties = list((self._definition.get("inputSchema") or {}).get("properties") or {})
    missing = [p for p in params if p not in properties]
    if missing:
        raise AssertionError(f"Tool '{self._name}' has no input properties {missing!r}; properties={properties!r}")
    return self

does_not_accept ¶

does_not_accept(params: list[str]) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def does_not_accept(self, params: list[str]) -> Self:
    properties = list((self._definition.get("inputSchema") or {}).get("properties") or {})
    present = [p for p in params if p in properties]
    if present:
        raise AssertionError(
            f"Tool '{self._name}' should not expose properties {present!r}; properties={properties!r}"
        )
    return self

has_output_schema ¶

has_output_schema() -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def has_output_schema(self) -> Self:
    if not self._definition.get("outputSchema"):
        raise AssertionError(f"Tool '{self._name}' has no outputSchema")
    return self

pyssertive.protocols.mcp.AssertableToolCall ¶

AssertableToolCall(
    *,
    name: str,
    result: dict[str, Any] | None,
    error: dict[str, Any] | None,
)

Source code in src/pyssertive/protocols/mcp/tools.py

def __init__(
    self,
    *,
    name: str,
    result: dict[str, Any] | None,
    error: dict[str, Any] | None,
) -> None:
    self._name = name
    self._result = result
    self._error = error

succeeds ¶

succeeds() -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def succeeds(self) -> Self:
    result = self._require_result()
    if result.get("isError") is True:
        raise AssertionError(f"Tool '{self._name}' reported isError=true; expected success")
    return self

returns_text ¶

returns_text(expected: str) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def returns_text(self, expected: str) -> Self:
    for block in self._content_blocks():
        if isinstance(block, dict) and block.get("type") == "text" and block.get("text") == expected:
            return self
    texts = [b.get("text") for b in self._content_blocks() if isinstance(b, dict) and b.get("type") == "text"]
    raise AssertionError(f"Tool '{self._name}' did not return text {expected!r}. Got text blocks: {texts!r}")

returns_text_containing ¶

returns_text_containing(substr: str) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def returns_text_containing(self, substr: str) -> Self:
    for block in self._content_blocks():
        if isinstance(block, dict) and block.get("type") == "text" and substr in (block.get("text") or ""):
            return self
    texts = [b.get("text") for b in self._content_blocks() if isinstance(b, dict) and b.get("type") == "text"]
    raise AssertionError(f"Tool '{self._name}' did not return any text block containing {substr!r}. Got: {texts!r}")

returns_image ¶

returns_image(*, mime_type: str | None = None) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def returns_image(self, *, mime_type: str | None = None) -> Self:
    for block in self._content_blocks():
        if (
            isinstance(block, dict)
            and block.get("type") == "image"
            and (mime_type is None or block.get("mimeType") == mime_type)
        ):
            return self
    raise AssertionError(
        f"Tool '{self._name}' did not return an image block"
        + (f" with mimeType {mime_type!r}" if mime_type else "")
    )

returns_content_count ¶

returns_content_count(expected: int) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def returns_content_count(self, expected: int) -> Self:
    actual = len(self._content_blocks())
    if actual != expected:
        raise AssertionError(f"Tool '{self._name}' returned {actual} content blocks, expected {expected}")
    return self

returns_structured ¶

returns_structured(expected: Any) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def returns_structured(self, expected: Any) -> Self:
    result = self._require_result()
    actual = result.get("structuredContent")
    if actual != expected:
        raise AssertionError(f"Tool '{self._name}' structuredContent: expected {expected!r}, got {actual!r}")
    return self

content ¶

content(
    index: int,
    callback: Callable[[AssertableContent], Any]
    | None = None,
) -> AssertableContent | Self

Source code in src/pyssertive/protocols/mcp/tools.py

def content(
    self,
    index: int,
    callback: Callable[[AssertableContent], Any] | None = None,
) -> AssertableContent | Self:
    typed = self._content_at(index)
    if callback is None:
        return typed
    callback(typed)
    return self

reports_tool_error ¶

reports_tool_error() -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def reports_tool_error(self) -> Self:
    result = self._require_result()
    if result.get("isError") is not True:
        raise AssertionError(f"Tool '{self._name}' did not report isError=true; got result={result!r}")
    return self

with_message_containing ¶

with_message_containing(substr: str) -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def with_message_containing(self, substr: str) -> Self:
    if self._error is not None:
        message = self._error.get("message") or ""
        if substr not in message:
            raise AssertionError(f"Tool '{self._name}' error message does not contain {substr!r}: {message!r}")
        return self
    result = self._require_result()
    content = result.get("content") or []
    for block in content:
        if isinstance(block, dict) and block.get("type") == "text" and substr in (block.get("text") or ""):
            return self
    raise AssertionError(f"Tool '{self._name}' tool-error message does not contain {substr!r}; content={content!r}")

is_rejected_as_unknown_tool ¶

is_rejected_as_unknown_tool() -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def is_rejected_as_unknown_tool(self) -> Self:
    error = self._require_error()
    code = error.get("code")
    message = (error.get("message") or "").lower()
    if code == -32601:
        return self
    if code == -32602 and ("unknown tool" in message or "not found" in message):
        return self
    raise AssertionError(
        f"Expected unknown-tool rejection for '{self._name}', got code={code} message={error.get('message')!r}"
    )

is_rejected_with_invalid_params ¶

is_rejected_with_invalid_params() -> Self

Source code in src/pyssertive/protocols/mcp/tools.py

def is_rejected_with_invalid_params(self) -> Self:
    error = self._require_error()
    if error.get("code") != -32602:
        raise AssertionError(f"Expected invalid-params (-32602) for '{self._name}', got code={error.get('code')}")
    return self