feat: server search/filter + test-all health column #30

Merged
the_og merged 2 commits from feat/server-list-ux into main 2026-07-08 11:56:37 -04:00
3 changed files with 315 additions and 2 deletions
+116 -2
View File
@@ -72,6 +72,12 @@ WARN = "#fbbf24"
STATUS_COLORS = {"ok": GOOD, "missing": BAD, "warn": WARN, "remote": "#60a5fa", "unknown": WARN}
STATUS_GLYPH = {"ok": "", "missing": "", "warn": "", "remote": "", "unknown": ""}
# Health dot (spawn-test outcome, see core.HealthStatus) shown per row in the
# server tables' "Health" column -- distinct from the PATH-dependency Status
# column above.
HEALTH_COLORS = {"ok": GOOD, "failed": BAD, "untested": MUTED}
HEALTH_GLYPH = {"ok": "", "failed": "", "untested": ""}
STYLESHEET = f"""
/* No font-family here on purpose: Qt already uses the native system UI font
on every platform (San Francisco / Segoe UI / desktop default). Naming
@@ -1416,6 +1422,12 @@ class MainWindow(QMainWindow):
self._focused_table = None
self._row_of_index: dict[int, tuple] = {}
self._undo_stack: list[list] = [] # each entry: snapshot of self.servers
self._filter_query = ""
self._health: dict[str, tuple[str, str]] = {} # server name -> (HealthStatus, summary)
self._test_all_queue: list[core.ServerEntry] = []
self._test_all_total = 0
self._test_all_done = 0
self._health_tester: SpawnTester | None = None
central = QWidget()
self.setCentralWidget(central)
@@ -1533,10 +1545,10 @@ class MainWindow(QMainWindow):
# --- left (server table) -------------------------------------------- #
def _make_server_table(self, object_name=None):
t = QTableWidget(0, 4)
t = QTableWidget(0, 5)
if object_name:
t.setObjectName(object_name)
t.setHorizontalHeaderLabels(["On", "Name", "Type", "Status"])
t.setHorizontalHeaderLabels(["On", "Name", "Type", "Status", "Health"])
t.verticalHeader().setVisible(False)
t.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows)
t.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
@@ -1546,6 +1558,7 @@ class MainWindow(QMainWindow):
h.setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch)
h.setSectionResizeMode(2, QHeaderView.ResizeMode.ResizeToContents)
h.setSectionResizeMode(3, QHeaderView.ResizeMode.ResizeToContents)
h.setSectionResizeMode(4, QHeaderView.ResizeMode.ResizeToContents)
t.itemSelectionChanged.connect(lambda tbl=t: self._on_selection(tbl))
t.itemChanged.connect(self._table_item_changed)
return t
@@ -1560,6 +1573,12 @@ class MainWindow(QMainWindow):
head.setObjectName("h1")
v.addWidget(head)
self.search_box = QLineEdit()
self.search_box.setPlaceholderText("Search servers by name, command, or url…")
self.search_box.setClearButtonEnabled(True)
self.search_box.textChanged.connect(self._on_search_changed)
v.addWidget(self.search_box)
# Active and Disabled sections live in a vertical splitter so the user
# can drag the divider instead of being stuck with a fixed-height
# disabled list.
@@ -1616,12 +1635,17 @@ class MainWindow(QMainWindow):
self.undo_btn = QPushButton("Undo")
self.undo_btn.setEnabled(False)
self.undo_btn.setToolTip("Undo last change (Ctrl+Z)")
self.test_all_btn = QPushButton("Test all")
self.test_all_btn.setToolTip(
"Spawn-test every enabled local server, one at a time, and fill in the Health column"
)
self.add_btn.clicked.connect(self.add_server)
self.dup_btn.clicked.connect(self.duplicate_server)
self.del_btn.clicked.connect(self.delete_server)
self.paste_btn.clicked.connect(self.paste_json)
self.copy_btn.clicked.connect(self.copy_to_menu)
self.undo_btn.clicked.connect(self._undo)
self.test_all_btn.clicked.connect(self._test_all_servers)
for b in (
self.add_btn,
self.dup_btn,
@@ -1629,6 +1653,7 @@ class MainWindow(QMainWindow):
self.paste_btn,
self.copy_btn,
self.undo_btn,
self.test_all_btn,
):
bar.addWidget(b)
# Ctrl+Z shortcut
@@ -1749,6 +1774,7 @@ class MainWindow(QMainWindow):
self.restart_btn.hide()
self._undo_stack.clear()
self.undo_btn.setEnabled(False)
self._health.clear() # health results are per-profile; a fresh load invalidates them
self._refresh_tables(select_index=0 if self.servers else -1)
self._update_status(saved=False)
if repaired:
@@ -1785,6 +1811,14 @@ class MainWindow(QMainWindow):
QColor(STATUS_COLORS.get(dep["status"], MUTED))
) # status stays colored even when off
table.setItem(r, 3, st)
health_status, health_summary = self._health.get(s.name, (core.HealthStatus.UNTESTED, ""))
if s.kind == "remote" and s.name not in self._health:
health_summary = "remote server — use “Test connection” in the editor"
health_item = QTableWidgetItem(HEALTH_GLYPH.get(health_status, ""))
health_item.setForeground(QColor(HEALTH_COLORS.get(health_status, MUTED)))
health_item.setTextAlignment(Qt.AlignmentFlag.AlignCenter)
health_item.setToolTip(health_summary or health_status)
table.setItem(r, 4, health_item)
self._row_of_index[master_idx] = (table, r)
def _refresh_tables(self, select_index=None):
@@ -1793,8 +1827,16 @@ class MainWindow(QMainWindow):
self._row_of_index = {}
self.active_table.setRowCount(0)
self.disabled_table.setRowCount(0)
query = self._filter_query
n_active = n_disabled = 0
total_active = total_disabled = 0
for i, s in enumerate(self.servers):
if s.enabled:
total_active += 1
else:
total_disabled += 1
if not core.server_matches_filter(s, query):
continue
if s.enabled:
self._add_row(self.active_table, i, s)
n_active += 1
@@ -1803,8 +1845,18 @@ class MainWindow(QMainWindow):
n_disabled += 1
self.active_table.setVisible(n_active > 0)
self.active_empty.setVisible(n_active == 0)
self.active_empty.setText(
"No matches."
if query.strip() and total_active and not n_active
else "No active servers. Add one, or Paste JSON."
)
self.disabled_table.setVisible(n_disabled > 0)
self.disabled_empty.setVisible(n_disabled == 0)
self.disabled_empty.setText(
"No matches."
if query.strip() and total_disabled and not n_disabled
else "Nothing disabled."
)
self._refresh_badges()
self._suppress_table = False
self._suppress_sel = False
@@ -1816,6 +1868,58 @@ class MainWindow(QMainWindow):
self._load_editor_from_selection()
self._validate()
# --- search / filter -------------------------------------------------- #
def _on_search_changed(self, text):
self._filter_query = text
cur = self._current_index()
self._refresh_tables(select_index=cur if cur >= 0 else None)
# --- test all (spawn-test every enabled local server) ---------------- #
def _test_all_servers(self):
targets = [s for s in self.servers if s.enabled and s.kind == "stdio"]
if not targets:
self.status.setText("No enabled local servers to test.")
return
self._test_all_queue = list(targets)
self._test_all_total = len(targets)
self._test_all_done = 0
self.test_all_btn.setEnabled(False)
self.test_all_btn.setText(f"Testing 0/{self._test_all_total}")
self._run_next_health_test()
def _run_next_health_test(self):
if not self._test_all_queue:
self.test_all_btn.setEnabled(True)
self.test_all_btn.setText("Test all")
self.status.setText(f"Tested {self._test_all_done} server(s).")
return
entry = self._test_all_queue.pop(0)
self._health_tester = SpawnTester(dict(entry.data), timeout=3.0)
self._health_tester.done.connect(
lambda result, name=entry.name: self._on_health_test_done(name, result)
)
self._health_tester.start()
def _on_health_test_done(self, name, result):
status, summary = core.health_from_spawn_result(result)
self._health[name] = (status, summary)
self._test_all_done += 1
self.test_all_btn.setText(f"Testing {self._test_all_done}/{self._test_all_total}")
self._update_health_cell(name, status, summary)
self._run_next_health_test()
def _update_health_cell(self, name, status, summary):
idx = next((i for i, s in enumerate(self.servers) if s.name == name), None)
if idx is None or idx not in self._row_of_index:
return
table, row = self._row_of_index[idx]
item = table.item(row, 4)
if item is None:
return
item.setText(HEALTH_GLYPH.get(status, ""))
item.setForeground(QColor(HEALTH_COLORS.get(status, MUTED)))
item.setToolTip(summary or status)
def _section_html(self, title, n, miss, warn):
base = f"{title} · {n}"
if miss:
@@ -1922,6 +2026,9 @@ class MainWindow(QMainWindow):
entry.data = self.editor.dump_data()
# The server stays in its section (enable state unchanged), so update
# its existing row in place rather than re-rendering.
# An edit invalidates any cached "Test all" result -- the server that
# was spawn-tested no longer matches what's on disk once saved.
self._health.pop(entry.name, None)
loc = self._row_of_index.get(idx)
if loc:
table, row = loc
@@ -1932,6 +2039,13 @@ class MainWindow(QMainWindow):
st = table.item(row, 3)
st.setText(f"{STATUS_GLYPH.get(dep['status'], '')} {dep['label']}")
st.setForeground(QColor(STATUS_COLORS.get(dep["status"], MUTED)))
health_item = table.item(row, 4)
if health_item is not None:
health_item.setText(HEALTH_GLYPH.get(core.HealthStatus.UNTESTED, ""))
health_item.setForeground(
QColor(HEALTH_COLORS.get(core.HealthStatus.UNTESTED, MUTED))
)
health_item.setToolTip("Not tested since last edit.")
self._suppress_table = False
self._refresh_badges()
self._mark_dirty()
+73
View File
@@ -1213,6 +1213,32 @@ def validate_servers(servers: list[ServerEntry]) -> list[str]:
return problems
# --------------------------------------------------------------------------- #
# Search / filter
# --------------------------------------------------------------------------- #
def server_matches_filter(entry: ServerEntry, query: str) -> bool:
"""
Case-insensitive substring match against a server's name, and its
command (stdio) or url (remote). An empty/whitespace-only query matches
everything -- that's what lets the search box double as "no filter".
"""
q = (query or "").strip().lower()
if not q:
return True
if q in entry.name.lower():
return True
if entry.kind == "remote":
haystack = str(entry.data.get("url", ""))
else:
haystack = str(entry.data.get("command", ""))
return q in haystack.lower()
def filter_servers(entries: list[ServerEntry], query: str) -> list[ServerEntry]:
"""Return only the entries that match `query` (see server_matches_filter)."""
return [e for e in entries if server_matches_filter(e, query)]
# --------------------------------------------------------------------------- #
# Dependency / PATH checking
# --------------------------------------------------------------------------- #
@@ -1682,6 +1708,53 @@ def spawn_test(data: dict, timeout: float = 3.0) -> dict:
}
# --------------------------------------------------------------------------- #
# Health status (maps a spawn_test() result to a simple tri-state for the
# server-list UI's per-row status dot; see "Test all" in bcc.py)
# --------------------------------------------------------------------------- #
class HealthStatus:
"""
Tri-state health for the server-list status dot. A plain class of string
constants -- not an Enum -- to match the plain-string status values used
elsewhere in this module (see check_dependency's 'status').
"""
UNTESTED = "untested"
OK = "ok"
FAILED = "failed"
def health_from_spawn_result(result: dict) -> tuple[str, str]:
"""
Map a spawn_test() result dict to (HealthStatus, short_summary) for the
server-list status column. Reuses spawn_test's own outcome classification
rather than re-deriving pass/fail from returncode/stderr:
outcome "ok" -> OK (server started and kept running)
outcome "not_applicable" -> UNTESTED (remote server, or no command set)
anything else -> FAILED (exited, crashed, or not found)
The summary is short enough for a table cell/tooltip; when the process
wrote to stderr before dying, its first line is appended for context.
"""
outcome = result.get("outcome", "")
detail = result.get("detail", "") or ""
stderr = (result.get("stderr") or "").strip()
if outcome == "ok":
return HealthStatus.OK, detail or "started"
if outcome == "not_applicable":
return HealthStatus.UNTESTED, detail or "not applicable"
# exited / crashed / not_found: the server didn't come up cleanly.
summary = detail or outcome
if stderr:
first_line = stderr.splitlines()[0].strip()
if first_line:
summary = f"{summary}{first_line}"
return HealthStatus.FAILED, summary
def test_remote(url: str, timeout: float = 5.0) -> tuple[bool, str]:
"""
Reachability check for a url-based MCP server. ANY HTTP response (even 4xx/5xx)
+126
View File
@@ -1200,3 +1200,129 @@ def test_fetch_latest_release_malformed_json_returns_none(monkeypatch):
urllib.request, "urlopen", lambda req, timeout=None: _FakeHTTPResponse(b"not json")
)
assert c.fetch_latest_release() is None
# --------------------------------------------------------------------------- #
# Server search / filter (#27)
# --------------------------------------------------------------------------- #
def test_filter_matches_name():
e = c.ServerEntry("brave-search", {"command": "npx", "args": ["-y", "@x/brave"]}, True)
assert c.server_matches_filter(e, "brave")
def test_filter_matches_stdio_command():
e = c.ServerEntry("filesystem", {"command": "uvx", "args": []}, True)
assert c.server_matches_filter(e, "uvx")
def test_filter_matches_remote_url():
e = c.ServerEntry("hosted", {"url": "https://mcp.example.com/sse"}, True)
assert c.server_matches_filter(e, "example.com")
def test_filter_is_case_insensitive():
e = c.ServerEntry("BraveSearch", {"command": "NPX", "args": []}, True)
assert c.server_matches_filter(e, "bravesearch")
assert c.server_matches_filter(e, "npx")
def test_filter_empty_query_matches_everything():
e = c.ServerEntry("anything", {"command": "node", "args": []}, True)
assert c.server_matches_filter(e, "")
assert c.server_matches_filter(e, " ")
def test_filter_no_match_returns_false():
e = c.ServerEntry("filesystem", {"command": "uvx", "args": []}, True)
assert not c.server_matches_filter(e, "nonexistent")
def test_filter_remote_query_does_not_match_stdio_command_field():
e = c.ServerEntry("hosted", {"url": "https://mcp.example.com/sse"}, True)
assert not c.server_matches_filter(e, "npx")
def test_filter_servers_returns_only_matches():
servers = [
c.ServerEntry("brave-search", {"command": "npx", "args": []}, True),
c.ServerEntry("filesystem", {"command": "uvx", "args": []}, True),
c.ServerEntry("hosted", {"url": "https://mcp.example.com/sse"}, True),
]
assert [s.name for s in c.filter_servers(servers, "brave")] == ["brave-search"]
assert [s.name for s in c.filter_servers(servers, "")] == [s.name for s in servers]
assert c.filter_servers(servers, "zzz-nope") == []
# --------------------------------------------------------------------------- #
# Health status mapping (#28)
# --------------------------------------------------------------------------- #
def test_health_from_spawn_result_ok():
result = {
"outcome": "ok",
"returncode": None,
"stderr": "",
"detail": "still running after 3s — server started successfully",
}
status, summary = c.health_from_spawn_result(result)
assert status == c.HealthStatus.OK
assert "started" in summary
def test_health_from_spawn_result_not_applicable_is_untested():
result = {
"outcome": "not_applicable",
"returncode": None,
"stderr": "",
"detail": "remote server",
}
status, summary = c.health_from_spawn_result(result)
assert status == c.HealthStatus.UNTESTED
assert summary == "remote server"
def test_health_from_spawn_result_crashed_is_failed():
result = {
"outcome": "crashed",
"returncode": 1,
"stderr": "Traceback: boom\nsecond line",
"detail": "process exited with code 1",
}
status, summary = c.health_from_spawn_result(result)
assert status == c.HealthStatus.FAILED
assert "process exited with code 1" in summary
assert "Traceback: boom" in summary
def test_health_from_spawn_result_exited_is_failed():
result = {
"outcome": "exited",
"returncode": 0,
"stderr": "",
"detail": "process exited cleanly (code 0) — unusual",
}
status, summary = c.health_from_spawn_result(result)
assert status == c.HealthStatus.FAILED
assert "exited cleanly" in summary
def test_health_from_spawn_result_not_found_is_failed():
result = {
"outcome": "not_found",
"returncode": None,
"stderr": "",
"detail": "command not found: totallybogus",
}
status, summary = c.health_from_spawn_result(result)
assert status == c.HealthStatus.FAILED
assert "not found" in summary
def test_health_from_spawn_result_failed_without_stderr_has_no_dash():
result = {
"outcome": "not_found",
"returncode": None,
"stderr": "",
"detail": "command not found: x",
}
_, summary = c.health_from_spawn_result(result)
assert "" not in summary