diff --git a/backend/apps/agents/core/mcp_preflight.py b/backend/apps/agents/core/mcp_preflight.py index fad2c739..5db88d3c 100644 --- a/backend/apps/agents/core/mcp_preflight.py +++ b/backend/apps/agents/core/mcp_preflight.py @@ -138,6 +138,20 @@ def _build_available_shortlist(settings) -> list[CuratedEntry]: ] +def offer_for_gated_server(server_name: str, settings) -> CuratedEntry | None: + """Mid-run a running agent may reach for a vetted MCP it isn't granted; this maps that + server to a one-click connect offer to SHOW the user. Suggest-only by construction: it + returns data to display, never an action that grants access, so it cannot widen the MCP + surface (activation stays behind MCPActivate + the dispatch gate). Returns None unless the + server is vetted AND inactive AND not dismissed, reusing the same filter as the preflight.""" + if not server_name or not isinstance(server_name, str): + return None + entry = next((e for e in _build_available_shortlist(settings) if e["id"] == server_name), None) + if entry is None: + return None + return {"id": entry["id"], "title": entry["title"], "description": entry["description"], "reason": ""} + + def _decorate(llm_suggestion: dict, available: list[CuratedEntry]) -> dict | None: """Expand an LLM-returned {id, reason} into the full frontend shape.""" entry = next((e for e in available if e["id"] == llm_suggestion["id"]), None) diff --git a/backend/tests/test_mcp_offer.py b/backend/tests/test_mcp_offer.py new file mode 100644 index 00000000..c494d9a5 --- /dev/null +++ b/backend/tests/test_mcp_offer.py @@ -0,0 +1,64 @@ +"""Gate-safety invariants for the Phase 2 in-task connect offer (offer_for_gated_server). + +The whole point of the offer is that it can ONLY ever suggest, never grant: it must surface a +vetted, inactive, not-dismissed MCP for the user to one-click-connect, and it must never carry +anything that could widen the MCP surface on its own. These tests make a bad offer state fail +loudly instead of shipping a silent gate bypass. +""" + +from types import SimpleNamespace + +import backend.apps.agents.core.mcp_preflight as pf +from backend.apps.agents.core.mcp_preflight import ( + CURATED_SHORTLIST, + offer_for_gated_server, +) + +VETTED = {e["id"] for e in CURATED_SHORTLIST} +OFFER_SHAPE = {"id", "title", "description", "reason"} + + +def _settings(dismissed=None): + return SimpleNamespace(dismissed_mcp_suggestions=dismissed or {}) + + +def test_offer_only_returns_vetted_inactive(monkeypatch): + monkeypatch.setattr(pf, "load_all_tools", lambda: []) # nothing enabled + s = _settings() + o = offer_for_gated_server("Google Workspace", s) + assert o is not None + assert o["id"] == "Google Workspace" + assert o["id"] in VETTED + + +def test_offer_rejects_unvetted_and_empty(monkeypatch): + monkeypatch.setattr(pf, "load_all_tools", lambda: []) + s = _settings() + assert offer_for_gated_server("NotAVettedServer", s) is None + assert offer_for_gated_server("", s) is None + assert offer_for_gated_server(None, s) is None # type: ignore[arg-type] + + +def test_offer_suppressed_when_dismissed(monkeypatch): + monkeypatch.setattr(pf, "load_all_tools", lambda: []) + s = _settings({"Google Workspace": "2026-01-01T00:00:00Z"}) + assert offer_for_gated_server("Google Workspace", s) is None + + +def test_offer_suppressed_when_already_active(monkeypatch): + monkeypatch.setattr( + pf, "load_all_tools", + lambda: [SimpleNamespace(name="Google Workspace", enabled=True)], + ) + s = _settings() + assert offer_for_gated_server("Google Workspace", s) is None + + +def test_offer_carries_no_activate_capability(monkeypatch): + # The security invariant: an offer is data to display, never an action that grants access. + monkeypatch.setattr(pf, "load_all_tools", lambda: []) + s = _settings() + for entry in CURATED_SHORTLIST: + o = offer_for_gated_server(entry["id"], s) + assert o is not None + assert set(o.keys()) == OFFER_SHAPE, f"offer for {entry['id']} grew an unexpected field"