Skip to content

Commit a3d9b69

Browse files
committed
mcp(feat[_utils]): Add _apply_filters() helper for QueryList filtering
why: MCP list tools need to expose libtmux's QueryList filtering via an optional dict parameter, requiring a bridge between MCP dict params and QueryList.filter(**kwargs). what: - Add _apply_filters() that validates operator keys against LOOKUP_NAME_MAP - Raise ToolError with valid operators list on invalid lookup operator - Short-circuit to direct serialization when filters is None/empty - Add 6 parametrized tests: none, empty, exact, no_match, invalid_op, contains
1 parent 118be15 commit a3d9b69

File tree

2 files changed

+149
-0
lines changed

2 files changed

+149
-0
lines changed

src/libtmux/mcp/_utils.py

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import typing as t
1313

1414
from libtmux import exc
15+
from libtmux._internal.query_list import LOOKUP_NAME_MAP
1516
from libtmux.server import Server
1617

1718
if t.TYPE_CHECKING:
@@ -273,6 +274,53 @@ def _resolve_pane(
273274
return panes[0]
274275

275276

277+
def _apply_filters(
278+
items: t.Any,
279+
filters: dict[str, str] | None,
280+
serializer: t.Callable[..., dict[str, t.Any]],
281+
) -> list[dict[str, t.Any]]:
282+
"""Apply QueryList filters and serialize results.
283+
284+
Parameters
285+
----------
286+
items : QueryList
287+
The QueryList of tmux objects to filter.
288+
filters : dict or None
289+
Django-style filter kwargs (e.g. ``{"session_name__contains": "dev"}``).
290+
If None or empty, all items are returned.
291+
serializer : callable
292+
Serializer function to convert each item to a dict.
293+
294+
Returns
295+
-------
296+
list[dict]
297+
Serialized list of matching items.
298+
299+
Raises
300+
------
301+
ToolError
302+
If a filter key uses an invalid lookup operator.
303+
"""
304+
if not filters:
305+
return [serializer(item) for item in items]
306+
307+
from fastmcp.exceptions import ToolError
308+
309+
valid_ops = sorted(LOOKUP_NAME_MAP.keys())
310+
for key in filters:
311+
if "__" in key:
312+
_field, op = key.rsplit("__", 1)
313+
if op not in LOOKUP_NAME_MAP:
314+
msg = (
315+
f"Invalid filter operator '{op}' in '{key}'. "
316+
f"Valid operators: {', '.join(valid_ops)}"
317+
)
318+
raise ToolError(msg)
319+
320+
filtered = items.filter(**filters)
321+
return [serializer(item) for item in filtered]
322+
323+
276324
def _serialize_session(session: Session) -> dict[str, t.Any]:
277325
"""Serialize a Session to a JSON-compatible dict.
278326

tests/mcp/test_utils.py

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,11 @@
55
import typing as t
66

77
import pytest
8+
from fastmcp.exceptions import ToolError
89

910
from libtmux import exc
1011
from libtmux.mcp._utils import (
12+
_apply_filters,
1113
_get_server,
1214
_invalidate_server,
1315
_resolve_pane,
@@ -144,3 +146,102 @@ def test_invalidate_server() -> None:
144146
assert len(_server_cache) == 1
145147
_invalidate_server(socket_name="test_inv")
146148
assert len(_server_cache) == 0
149+
150+
151+
class ApplyFiltersFixture(t.NamedTuple):
152+
"""Test fixture for _apply_filters."""
153+
154+
test_id: str
155+
filters: dict[str, str] | None
156+
expected_count: int | None # None = don't check exact count
157+
expect_error: bool
158+
error_match: str | None
159+
160+
161+
APPLY_FILTERS_FIXTURES: list[ApplyFiltersFixture] = [
162+
ApplyFiltersFixture(
163+
test_id="none_returns_all",
164+
filters=None,
165+
expected_count=None,
166+
expect_error=False,
167+
error_match=None,
168+
),
169+
ApplyFiltersFixture(
170+
test_id="empty_dict_returns_all",
171+
filters={},
172+
expected_count=None,
173+
expect_error=False,
174+
error_match=None,
175+
),
176+
ApplyFiltersFixture(
177+
test_id="exact_match",
178+
filters={"session_name": "<session_name>"},
179+
expected_count=1,
180+
expect_error=False,
181+
error_match=None,
182+
),
183+
ApplyFiltersFixture(
184+
test_id="no_match_returns_empty",
185+
filters={"session_name": "nonexistent_xyz_999"},
186+
expected_count=0,
187+
expect_error=False,
188+
error_match=None,
189+
),
190+
ApplyFiltersFixture(
191+
test_id="invalid_operator",
192+
filters={"session_name__badop": "test"},
193+
expected_count=None,
194+
expect_error=True,
195+
error_match="Invalid filter operator",
196+
),
197+
ApplyFiltersFixture(
198+
test_id="contains_operator",
199+
filters={"session_name__contains": "<partial>"},
200+
expected_count=1,
201+
expect_error=False,
202+
error_match=None,
203+
),
204+
]
205+
206+
207+
@pytest.mark.parametrize(
208+
ApplyFiltersFixture._fields,
209+
APPLY_FILTERS_FIXTURES,
210+
ids=[f.test_id for f in APPLY_FILTERS_FIXTURES],
211+
)
212+
def test_apply_filters(
213+
mcp_server: Server,
214+
mcp_session: Session,
215+
test_id: str,
216+
filters: dict[str, str] | None,
217+
expected_count: int | None,
218+
expect_error: bool,
219+
error_match: str | None,
220+
) -> None:
221+
"""_apply_filters bridges dict params to QueryList.filter()."""
222+
# Substitute placeholders with real session name
223+
if filters is not None:
224+
session_name = mcp_session.session_name
225+
assert session_name is not None
226+
resolved: dict[str, str] = {}
227+
for k, v in filters.items():
228+
if v == "<session_name>":
229+
resolved[k] = session_name
230+
elif v == "<partial>":
231+
resolved[k] = session_name[:4]
232+
else:
233+
resolved[k] = v
234+
filters = resolved
235+
236+
sessions = mcp_server.sessions
237+
238+
if expect_error:
239+
with pytest.raises(ToolError, match=error_match):
240+
_apply_filters(sessions, filters, _serialize_session)
241+
else:
242+
result = _apply_filters(sessions, filters, _serialize_session)
243+
assert isinstance(result, list)
244+
if expected_count is not None:
245+
assert len(result) == expected_count
246+
else:
247+
assert len(result) >= 1

0 commit comments

Comments
 (0)