diff --git a/README.md b/README.md index ae2309c..7914c65 100644 --- a/README.md +++ b/README.md @@ -492,6 +492,14 @@ 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.does_not_accept(["internal_user_id"]) +) +``` + #### 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..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") @@ -79,6 +88,12 @@ 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: + if isinstance(tool, dict): + 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..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") @@ -103,3 +124,41 @@ 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) + + +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