Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
15 changes: 15 additions & 0 deletions src/pyssertive/protocols/mcp/tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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")
Expand Down
59 changes: 59 additions & 0 deletions tests/protocols/mcp/test_tools_list.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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