From e362ebba31d58c421d0ddda9ff3480126557c660 Mon Sep 17 00:00:00 2001 From: Unay Santisteban Date: Thu, 14 May 2026 16:14:44 +1000 Subject: [PATCH 1/2] feat(mcp): #30 add every_tool for catalog-wide tool assertions --- README.md | 6 ++++++ src/pyssertive/protocols/mcp/tools.py | 5 +++++ tests/protocols/mcp/test_tools_list.py | 27 ++++++++++++++++++++++++++ 3 files changed, 38 insertions(+) diff --git a/README.md b/README.md index ae2309c..e8e6619 100644 --- a/README.md +++ b/README.md @@ -492,6 +492,12 @@ AssertableMCP(payload).lists_tools()\ )) ``` +Catalog-wide invariants — apply the same assertion to every tool without enumerating names. Useful when a server rewrites its tool schema per caller (auth scopes, feature flags): + +```python +AssertableMCP(payload).lists_tools().every_tool(lambda t: t.documented()) +``` + #### Building requests with `MessageBuilder` `MessageBuilder` constructs MCP JSON-RPC messages with native MCP vocabulary on top of any transport-level `RequestBuilder`. Inject `HttpxRequestBuilder` for FastAPI/Starlette/FastMCP testing, or `DjangoRequestBuilder` for Django-hosted MCP servers — the output of `.build()` matches whichever you inject. diff --git a/src/pyssertive/protocols/mcp/tools.py b/src/pyssertive/protocols/mcp/tools.py index c3b3761..5579523 100644 --- a/src/pyssertive/protocols/mcp/tools.py +++ b/src/pyssertive/protocols/mcp/tools.py @@ -79,6 +79,11 @@ def does_not_contain_tool(self, name: str) -> Self: raise AssertionError(f"Tool '{name}' should not be in tools list, but it was found") return self + def every_tool(self, callback: Callable[[AssertableToolDef], Any]) -> Self: + for tool in self._tools: + callback(AssertableToolDef(tool)) + return self + def has_more_pages(self) -> Self: if not self._result.get("nextCursor"): raise AssertionError("Expected tools/list response to advertise a nextCursor") diff --git a/tests/protocols/mcp/test_tools_list.py b/tests/protocols/mcp/test_tools_list.py index 59f119a..4d4a583 100644 --- a/tests/protocols/mcp/test_tools_list.py +++ b/tests/protocols/mcp/test_tools_list.py @@ -103,3 +103,30 @@ def test_lists_tools_has_more_pages_should_pass_when_next_cursor_present(): def test_lists_tools_has_more_pages_should_raise_when_next_cursor_absent(): with pytest.raises(AssertionError, match="nextCursor"): AssertableMCP(_list_response([{"name": "a"}])).lists_tools().has_more_pages() + + +def test_every_tool_should_apply_callback_to_each_tool(): + payload = _list_response([{"name": "a"}, {"name": "b"}, {"name": "c"}]) + call_count = [] + AssertableMCP(payload).lists_tools().every_tool(lambda t: call_count.append(1)) + assert len(call_count) == 3 + + +def test_every_tool_should_raise_with_tool_name_when_callback_fails(): + payload = _list_response( + [ + {"name": "a", "description": "doc a"}, + {"name": "b"}, + ] + ) + with pytest.raises(AssertionError, match="Tool 'b' has no description"): + AssertableMCP(payload).lists_tools().every_tool(lambda t: t.documented()) + + +def test_every_tool_should_pass_silently_when_tools_list_is_empty(): + AssertableMCP(_list_response([])).lists_tools().every_tool(lambda t: t.documented()) + + +def test_every_tool_should_return_self_for_chaining(): + payload = _list_response([{"name": "a", "description": "doc a"}]) + AssertableMCP(payload).lists_tools().every_tool(lambda t: t.documented()).with_count(1) From c089af5135cc885551e4dab766609cff967ca6b9 Mon Sep 17 00:00:00 2001 From: Unay Santisteban Date: Thu, 14 May 2026 16:31:34 +1000 Subject: [PATCH 2/2] feat(mcp): #30 add does_not_accept and harden every_tool against non-dict items --- README.md | 4 +++- src/pyssertive/protocols/mcp/tools.py | 12 +++++++++- tests/protocols/mcp/test_tools_list.py | 32 ++++++++++++++++++++++++++ 3 files changed, 46 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index e8e6619..7914c65 100644 --- a/README.md +++ b/README.md @@ -495,7 +495,9 @@ AssertableMCP(payload).lists_tools()\ Catalog-wide invariants — apply the same assertion to every tool without enumerating names. Useful when a server rewrites its tool schema per caller (auth scopes, feature flags): ```python -AssertableMCP(payload).lists_tools().every_tool(lambda t: t.documented()) +AssertableMCP(payload).lists_tools().every_tool( + lambda t: t.does_not_accept(["internal_user_id"]) +) ``` #### Building requests with `MessageBuilder` diff --git a/src/pyssertive/protocols/mcp/tools.py b/src/pyssertive/protocols/mcp/tools.py index 5579523..3497a0f 100644 --- a/src/pyssertive/protocols/mcp/tools.py +++ b/src/pyssertive/protocols/mcp/tools.py @@ -41,6 +41,15 @@ def accepts_optional(self, params: list[str]) -> Self: raise AssertionError(f"Tool '{self._name}' has no input properties {missing!r}; properties={properties!r}") return self + 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 + def has_output_schema(self) -> Self: if not self._definition.get("outputSchema"): raise AssertionError(f"Tool '{self._name}' has no outputSchema") @@ -81,7 +90,8 @@ def does_not_contain_tool(self, name: str) -> Self: def every_tool(self, callback: Callable[[AssertableToolDef], Any]) -> Self: for tool in self._tools: - callback(AssertableToolDef(tool)) + if isinstance(tool, dict): + callback(AssertableToolDef(tool)) return self def has_more_pages(self) -> Self: diff --git a/tests/protocols/mcp/test_tools_list.py b/tests/protocols/mcp/test_tools_list.py index 4d4a583..65d634f 100644 --- a/tests/protocols/mcp/test_tools_list.py +++ b/tests/protocols/mcp/test_tools_list.py @@ -84,6 +84,27 @@ def test_tool_def_has_output_schema_should_raise_when_field_absent(): AssertableMCP(payload).lists_tools().contains_tool("x", lambda t: t.has_output_schema()) +def test_tool_def_does_not_accept_should_pass_when_params_absent(): + payload = _list_response([{"name": "x", "inputSchema": {"properties": {"a": {}}}}]) + AssertableMCP(payload).lists_tools().contains_tool("x", lambda t: t.does_not_accept(["b", "c"])) + + +def test_tool_def_does_not_accept_should_pass_when_no_input_schema(): + payload = _list_response([{"name": "x"}]) + AssertableMCP(payload).lists_tools().contains_tool("x", lambda t: t.does_not_accept(["any"])) + + +def test_tool_def_does_not_accept_should_raise_when_param_present(): + payload = _list_response([{"name": "x", "inputSchema": {"properties": {"internal": {}}}}]) + with pytest.raises(AssertionError, match=r"Tool 'x' should not expose properties \['internal'\]"): + AssertableMCP(payload).lists_tools().contains_tool("x", lambda t: t.does_not_accept(["internal"])) + + +def test_tool_def_does_not_accept_should_return_self_for_chaining(): + payload = _list_response([{"name": "x", "description": "doc", "inputSchema": {"properties": {"a": {}}}}]) + AssertableMCP(payload).lists_tools().contains_tool("x", lambda t: t.does_not_accept(["forbidden"]).documented()) + + def test_does_not_contain_tool_should_pass_when_tool_absent(): payload = _list_response([{"name": "send_email"}]) AssertableMCP(payload).lists_tools().does_not_contain_tool("get_weather") @@ -130,3 +151,14 @@ def test_every_tool_should_pass_silently_when_tools_list_is_empty(): def test_every_tool_should_return_self_for_chaining(): payload = _list_response([{"name": "a", "description": "doc a"}]) AssertableMCP(payload).lists_tools().every_tool(lambda t: t.documented()).with_count(1) + + +def test_every_tool_should_skip_non_dict_items(): + payload = { + "jsonrpc": "2.0", + "id": 1, + "result": {"tools": [{"name": "a"}, "not a dict", {"name": "b"}]}, + } + call_count = [] + AssertableMCP(payload).lists_tools().every_tool(lambda t: call_count.append(1)) + assert len(call_count) == 2