diff --git a/backend/druks/mcp/models.py b/backend/druks/mcp/models.py
index e53c1c88..303d6e48 100644
--- a/backend/druks/mcp/models.py
+++ b/backend/druks/mcp/models.py
@@ -40,7 +40,7 @@ class McpServer(Base, Uuid7Pk):
 
     @classmethod
     def list_all(cls) -> list["McpServer"]:
-        # The raw overlay rows — not the merged registry view (list_resolved).
+        # The raw overlay rows — not the merged registry view (get_resolved).
         return list(db_session().execute(select(cls).order_by(cls.name)).scalars())
 
     @classmethod
@@ -48,46 +48,43 @@ def get_by_name(cls, name: str) -> "McpServer | None":
         return db_session().execute(select(cls).where(cls.name == name)).scalar_one_or_none()
 
     @classmethod
-    def list_resolved(cls) -> list[dict]:
-        # The full view the API reads and delivery resolves from: each built-in
-        # definition (url + auth from the registry) overlaid with its operator
-        # row's enable choice and secrets, then any fully custom rows.
+    def get_resolved(cls) -> dict[str, dict]:
+        # The full view the API reads and delivery resolves from, keyed by
+        # name: each built-in definition (url + auth from the registry)
+        # overlaid with its operator row's enable choice and secrets, then any
+        # fully custom rows.
         rows = {server.name: server for server in cls.list_all()}
-        servers: list[dict] = []
+        servers: dict[str, dict] = {}
         for definition in mcp_servers.all():
             row = rows.pop(definition["name"], None)
-            servers.append(
-                {
-                    "name": definition["name"],
-                    "url": definition["url"],
-                    "token_source": definition["token_source"],
-                    "source_env_var": definition["source_env_var"],
-                    "is_enabled": row.is_enabled if row else definition["enabled"],
-                    "token": row.token if row else Secret(b"", ""),
-                    "headers": row.headers if row else {},
-                    "secret_headers": row.secret_headers if row else {},
-                    "builtin": True,
-                }
-            )
+            servers[definition["name"]] = {
+                "name": definition["name"],
+                "url": definition["url"],
+                "token_source": definition["token_source"],
+                "source_env_var": definition["source_env_var"],
+                "is_enabled": row.is_enabled if row else definition["enabled"],
+                "token": row.token if row else Secret(b"", ""),
+                "headers": row.headers if row else {},
+                "secret_headers": row.secret_headers if row else {},
+                "builtin": True,
+            }
         for row in rows.values():
-            servers.append(
-                {
-                    "name": row.name,
-                    "url": row.url,
-                    "token_source": row.token_source,
-                    "source_env_var": "",
-                    "is_enabled": row.is_enabled,
-                    "token": row.token,
-                    "headers": row.headers,
-                    "secret_headers": row.secret_headers,
-                    "builtin": False,
-                }
-            )
+            servers[row.name] = {
+                "name": row.name,
+                "url": row.url,
+                "token_source": row.token_source,
+                "source_env_var": "",
+                "is_enabled": row.is_enabled,
+                "token": row.token,
+                "headers": row.headers,
+                "secret_headers": row.secret_headers,
+                "builtin": False,
+            }
         # has_token = nothing blocks this server's auth at delivery, read from
         # wherever its source keeps the secret: druks' env for an env-sourced
         # server, a stored grant for a connected one, the stored token for a
         # static one; a bearerless server has none to miss.
-        for server in servers:
+        for server in servers.values():
             source = server["token_source"]
             if not source:
                 server["has_token"] = True
@@ -102,7 +99,7 @@ def list_resolved(cls) -> list[dict]:
     @classmethod
     def list_enabled(cls) -> list[dict]:
         # The enabled subset — what a run delivers and the settings UI shows active.
-        return [server for server in cls.list_resolved() if server["is_enabled"]]
+        return [server for server in cls.get_resolved().values() if server["is_enabled"]]
 
     @classmethod
     def set_enabled(cls, name: str, is_enabled: bool) -> bool:
diff --git a/backend/druks/mcp/registry.py b/backend/druks/mcp/registry.py
index 9e33ceea..14a35f33 100644
--- a/backend/druks/mcp/registry.py
+++ b/backend/druks/mcp/registry.py
@@ -62,11 +62,12 @@ async def search_registry(query: str) -> list[dict]:
     return entries
 
 
-def resolve_candidates(entries: list[dict], pins: dict[str, str]) -> list[dict]:
+def resolve_candidates(entries: list[dict], pins: dict[str, str]) -> dict[str, dict]:
     """One installable candidate per entry reachable over streamable HTTP —
     from its own declared remote, or from a pinned official url the registry
-    omits; stdio/oci-only entries drop out. ``official`` is the trust badge:
-    the publisher provably owns the remote's domain, or a pin vouches."""
+    omits; stdio/oci-only entries drop out. Keyed by registry name, official
+    candidates first. ``official`` is the trust badge: the publisher provably
+    owns the remote's domain, or a pin vouches."""
     # A url pin lifts exactly one entry among those deriving its name. The pin
     # itself supplies the url, so this choice only picks display text; the
     # tiebreak just needs determinism: prefer the publisher whose own name
@@ -128,4 +129,4 @@ def resolve_candidates(entries: list[dict], pins: dict[str, str]) -> list[dict]:
             }
         )
     candidates.sort(key=lambda candidate: (not candidate["official"], candidate["name"]))
-    return candidates
+    return {candidate["registry_name"]: candidate for candidate in candidates}
diff --git a/backend/druks/mcp/routes.py b/backend/druks/mcp/routes.py
index 89e49d1f..4b728e68 100644
--- a/backend/druks/mcp/routes.py
+++ b/backend/druks/mcp/routes.py
@@ -1,14 +1,22 @@
+import json
+
 from fastapi import APIRouter, Body, HTTPException, Request
 from fastapi.responses import HTMLResponse
 
 from druks.extensions.registry import mcp_servers
-from druks.mcp import oauth
+from druks.mcp import oauth, registry
 from druks.mcp.enums import TokenSource
-from druks.mcp.exceptions import InvalidServerNameError, OauthConnectError
+from druks.mcp.exceptions import (
+    InvalidServerNameError,
+    OauthConnectError,
+    RegistryUnavailableError,
+)
 from druks.mcp.models import McpOauthGrant, McpServer
 from druks.mcp.schemas import (
     ConnectMcpServerResponse,
     CreateMcpServerRequest,
+    InstallMcpServerRequest,
+    McpRegistryCandidateResponse,
     McpServerResponse,
 )
 
@@ -16,14 +24,27 @@
 
 
 def _response(name: str) -> McpServerResponse:
-    return McpServerResponse.model_validate(
-        next(s for s in McpServer.list_resolved() if s["name"] == name)
-    )
+    return McpServerResponse.model_validate(McpServer.get_resolved()[name])
 
 
 @router.get("", response_model=list[McpServerResponse])
 async def list_mcp_servers() -> list[McpServerResponse]:
-    return [McpServerResponse.model_validate(server) for server in McpServer.list_resolved()]
+    return [
+        McpServerResponse.model_validate(server) for server in McpServer.get_resolved().values()
+    ]
+
+
+@router.get("/registry", response_model=list[McpRegistryCandidateResponse])
+async def search_mcp_registry(query: str, request: Request) -> list[McpRegistryCandidateResponse]:
+    pins = json.loads(request.app.state.settings.mcp_trusted_path.read_text())
+    try:
+        entries = await registry.search_registry(query)
+    except RegistryUnavailableError as error:
+        raise HTTPException(status_code=502, detail=str(error)) from error
+    candidates = registry.resolve_candidates(entries, pins)
+    return [
+        McpRegistryCandidateResponse.model_validate(candidate) for candidate in candidates.values()
+    ]
 
 
 @router.post("", response_model=McpServerResponse)
@@ -53,6 +74,70 @@ async def add_mcp_server(body: CreateMcpServerRequest) -> McpServerResponse:
     return _response(body.name)
 
 
+@router.post("/registry", response_model=McpServerResponse)
+async def install_mcp_server(body: InstallMcpServerRequest, request: Request) -> McpServerResponse:
+    if body.name in mcp_servers:
+        raise HTTPException(
+            status_code=409,
+            detail=f"MCP server {body.name!r} is built-in; configure it instead of adding it.",
+        )
+    if McpServer.get_by_name(body.name):
+        raise HTTPException(
+            status_code=409, detail=f"MCP server {body.name!r} already exists; remove it first."
+        )
+    # url, auth shape and header secrecy come from the re-resolved registry
+    # entry, never the client.
+    pins = json.loads(request.app.state.settings.mcp_trusted_path.read_text())
+    try:
+        entries = await registry.search_registry(body.registry)
+    except RegistryUnavailableError as error:
+        raise HTTPException(status_code=502, detail=str(error)) from error
+    candidate = registry.resolve_candidates(entries, pins).get(body.registry)
+    if not candidate:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Registry entry {body.registry!r} is not installable over HTTP.",
+        )
+    declared = {spec["name"] for spec in candidate["headers"]}
+    unknown = sorted(set(body.headers) - declared)
+    if unknown:
+        raise HTTPException(
+            status_code=422,
+            detail=f"{body.registry!r} declares no header(s): {', '.join(unknown)}.",
+        )
+    filled = {}
+    for header, value in body.headers.items():
+        if stripped := value.strip():
+            filled[header] = stripped
+    required = {spec["name"] for spec in candidate["headers"] if spec.get("isRequired")}
+    missing = sorted(required - set(filled))
+    if missing:
+        raise HTTPException(
+            status_code=422, detail=f"Missing required header value(s): {', '.join(missing)}."
+        )
+    secret = {spec["name"] for spec in candidate["headers"] if spec.get("isSecret")}
+    if secret:
+        # A secret declared header carries the auth itself — no bearer.
+        token_source = ""
+        is_enabled = True
+    else:
+        # OAuth: ships dark until its Connect lands.
+        token_source = TokenSource.OAUTH
+        is_enabled = False
+    try:
+        McpServer.create(
+            name=body.name,
+            url=candidate["url"],
+            token_source=token_source,
+            headers={h: v for h, v in filled.items() if h not in secret},
+            secret_headers={h: v for h, v in filled.items() if h in secret},
+            is_enabled=is_enabled,
+        )
+    except InvalidServerNameError as error:
+        raise HTTPException(status_code=422, detail=str(error)) from error
+    return _response(body.name)
+
+
 @router.patch("/{name}", response_model=McpServerResponse)
 async def set_mcp_server_enabled(
     name: str, is_enabled: bool = Body(embed=True)
@@ -74,12 +159,16 @@ async def remove_mcp_server(name: str) -> None:
     if not server:
         raise HTTPException(status_code=404, detail=f"MCP server {name!r} not found")
     server.delete()
+    if grant := McpOauthGrant.get_by_server(name):
+        # An orphan grant would revive as this name's credential on re-add.
+        grant.delete()
+        await oauth.evict_access_token(name)
 
 
 @router.post("/{name}/connect", response_model=ConnectMcpServerResponse)
 async def connect_mcp_server(name: str, request: Request) -> ConnectMcpServerResponse:
-    definition = mcp_servers.get(name)
-    if not definition or definition["token_source"] != TokenSource.OAUTH:
+    server = McpServer.get_resolved().get(name)
+    if not server or server["token_source"] != TokenSource.OAUTH:
         raise HTTPException(status_code=404, detail=f"MCP server {name!r} is not an OAuth server.")
     endpoint = request.app.state.settings.endpoint
     if not endpoint:
@@ -91,7 +180,7 @@ async def connect_mcp_server(name: str, request: Request) -> ConnectMcpServerRes
             "at, to connect OAuth MCP servers.",
         )
     try:
-        authorization_url = await oauth.begin_connect(name, definition["url"], endpoint)
+        authorization_url = await oauth.begin_connect(name, server["url"], endpoint)
     except OauthConnectError as error:
         raise HTTPException(status_code=502, detail=str(error)) from error
     return ConnectMcpServerResponse(authorization_url=authorization_url)
diff --git a/backend/druks/mcp/schemas.py b/backend/druks/mcp/schemas.py
index 778f5aa6..fac34388 100644
--- a/backend/druks/mcp/schemas.py
+++ b/backend/druks/mcp/schemas.py
@@ -8,7 +8,7 @@
 
 
 class McpServerResponse(BaseResponse):
-    # A pure projection of one ``McpServer.list_resolved()`` item — the dict's
+    # A pure projection of one ``McpServer.get_resolved()`` item — the dict's
     # ``token`` is not a field here, so the secret can't serialize (and it
     # arrives as a Secret, redacted even if it did).
     name: str
@@ -29,12 +29,28 @@ class ConnectMcpServerResponse(BaseResponse):
     authorization_url: str
 
 
+class McpRegistryCandidateResponse(BaseResponse):
+    name: str
+    registry_name: str
+    description: str
+    url: str
+    official: bool
+    # The remote's declared inputs, verbatim — the registry owns their shape.
+    headers: list[dict]
+
+
 class CreateMcpServerRequest(BaseModel):
     name: str
     url: str
     token: str = ""
 
 
+class InstallMcpServerRequest(BaseModel):
+    name: str
+    registry: str
+    headers: dict[str, str] = {}
+
+
 # The catalog file is operator input, so its entries parse through a
 # discriminated union — each auth strategy carries exactly its own fields, and
 # an unknown key is rejected (a stray key in a catalog is a typo, not
diff --git a/backend/tests/test_mcp_registry.py b/backend/tests/test_mcp_registry.py
index e2310b7c..8972a4e1 100644
--- a/backend/tests/test_mcp_registry.py
+++ b/backend/tests/test_mcp_registry.py
@@ -2,12 +2,15 @@
 
 import httpx
 import pytest
+from conftest import configure_app_for_test, make_settings
 from druks.mcp import registry
 from druks.mcp.constants import REGISTRY_SEARCH_CACHE_PREFIX
 from druks.mcp.exceptions import RegistryUnavailableError
+from druks.mcp.models import McpOauthGrant, McpServer
 from druks.mcp.registry import derive_server_name, resolve_candidates, search_registry
 from druks.redis import get_client
 from druks.settings import PACKAGED_MCP_TRUSTED
+from fastapi.testclient import TestClient
 
 # Canned latest-version entries, shaped like the live /v0/servers?search=…
 # payload (verified against the registry 2026-07-13): grafana publishes a
@@ -53,7 +56,7 @@ def _entry(name, *, description="aggregated wrapper", remotes=None):
             "description": description,
             "version": "1.0.0",
             "packages": [{"registryType": "pypi", "transport": {"type": "stdio"}}],
-            **({"remotes": remotes} if remotes is not None else {}),
+            **({"remotes": remotes} if remotes else {}),
         },
         "_meta": {"io.modelcontextprotocol.registry/official": {"isLatest": True}},
     }
@@ -71,11 +74,10 @@ def test_publisher_pin_marks_grafana_official_with_the_registry_url():
     # stays the live registry value.
     candidates = resolve_candidates([_GRAFANA], _PINS)
 
-    assert [c["name"] for c in candidates] == ["grafana"]
-    grafana = candidates[0]
+    grafana = candidates["io.github.grafana/mcp-grafana"]
+    assert grafana["name"] == "grafana"
     assert grafana["official"] is True
     assert grafana["url"] == "https://mcp.grafana.com/mcp"
-    assert grafana["registry_name"] == "io.github.grafana/mcp-grafana"
     assert grafana["description"].startswith("An MCP server")
 
 
@@ -84,7 +86,7 @@ def test_url_pin_lifts_sentry_whose_registry_entry_has_no_remote():
     # text still comes from the registry entry.
     candidates = resolve_candidates([_SENTRY], _PINS)
 
-    sentry = next(c for c in candidates if c["name"] == "sentry")
+    sentry = candidates["io.github.getsentry/sentry-mcp"]
     assert sentry["official"] is True
     assert sentry["url"] == "https://mcp.sentry.dev/mcp"
     assert "error monitoring" in sentry["description"]
@@ -99,10 +101,8 @@ def test_url_pin_attaches_to_the_product_publisher_not_the_aggregator():
 
     candidates = resolve_candidates(entries, _PINS)
 
-    sentries = [c for c in candidates if c["name"] == "sentry"]
-    assert len(sentries) == 1
-    assert sentries[0]["registry_name"] == "io.github.getsentry/sentry-mcp"
-    assert "error monitoring" in sentries[0]["description"]
+    assert list(candidates) == ["io.github.getsentry/sentry-mcp"]
+    assert "error monitoring" in candidates["io.github.getsentry/sentry-mcp"]["description"]
 
 
 def test_domain_ownership_rule_needs_no_pin():
@@ -117,7 +117,7 @@ def test_domain_ownership_rule_needs_no_pin():
 
     candidates = resolve_candidates(entries, {})
 
-    assert candidates[0]["official"] is True
+    assert candidates["com.cloudflare/browser"]["official"] is True
 
 
 def test_unpinned_unmatched_remote_is_community():
@@ -130,14 +130,14 @@ def test_unpinned_unmatched_remote_is_community():
 
     candidates = resolve_candidates(entries, _PINS)
 
-    assert [c["official"] for c in candidates] == [False]
+    assert candidates["io.github.someone/grafana-tools"]["official"] is False
 
 
 def test_stdio_only_entries_are_dropped():
     # No streamable-http remote and no url pin — not installable, not shown.
     candidates = resolve_candidates([_entry("com.mcparmory/grafana")], _PINS)
 
-    assert candidates == []
+    assert candidates == {}
 
 
 def test_official_candidates_sort_first():
@@ -148,7 +148,7 @@ def test_official_candidates_sort_first():
 
     candidates = resolve_candidates([community, _GRAFANA], _PINS)
 
-    assert [c["name"] for c in candidates] == ["grafana", "aardvark"]
+    assert [c["name"] for c in candidates.values()] == ["grafana", "aardvark"]
 
 
 def test_packaged_pins_resolve_grafana_and_sentry():
@@ -158,11 +158,11 @@ def test_packaged_pins_resolve_grafana_and_sentry():
 
     candidates = resolve_candidates([_GRAFANA, _SENTRY], pins)
 
-    assert [(c["name"], c["official"]) for c in candidates] == [
+    assert [(c["name"], c["official"]) for c in candidates.values()] == [
         ("grafana", True),
         ("sentry", True),
     ]
-    assert next(c for c in candidates if c["name"] == "sentry")["url"] == pins["sentry"]
+    assert candidates["io.github.getsentry/sentry-mcp"]["url"] == pins["sentry"]
 
 
 # --- declared inputs: the form spec ---------------------------------------
@@ -173,7 +173,7 @@ def test_declared_header_inputs_pass_through_verbatim():
     # response model owns their optionality, not the resolver.
     candidates = resolve_candidates([_GRAFANA], _PINS)
 
-    assert candidates[0]["headers"] == [_GRAFANA_HEADER]
+    assert candidates["io.github.grafana/mcp-grafana"]["headers"] == [_GRAFANA_HEADER]
 
 
 # --- the druks-side name ---------------------------------------------------
@@ -254,7 +254,207 @@ async def test_search_registry_result_feeds_the_resolver(monkeypatch):
 
     candidates = resolve_candidates(await search_registry("observability"), _PINS)
 
-    assert [(c["name"], c["official"]) for c in candidates] == [
+    assert [(c["name"], c["official"]) for c in candidates.values()] == [
         ("grafana", True),
         ("sentry", True),
     ]
+
+
+# --- the install API --------------------------------------------------------
+
+_ACME_ENTRY = _entry(
+    "com.acme/observer",
+    description="Observability for agents",
+    remotes=[
+        {
+            "type": "streamable-http",
+            "url": "https://mcp.acme.com/mcp",
+            "headers": [
+                {
+                    "name": "X-Api-Key",
+                    "description": "Acme API key",
+                    "isSecret": True,
+                    "isRequired": True,
+                },
+                {"name": "X-Region", "description": "Acme region"},
+            ],
+        }
+    ],
+)
+
+
+def _client_with_registry(tmp_path, monkeypatch, *entries):
+    payload = {"servers": list(entries)}
+    monkeypatch.setattr(
+        registry,
+        "_http",
+        _client_returning(lambda _r: httpx.Response(200, json=payload)),
+    )
+    app = configure_app_for_test(settings=make_settings(tmp_path, endpoint="http://druks.test"))
+    return TestClient(app)
+
+
+def test_registry_search_route_projects_resolved_candidates(tmp_path, monkeypatch, db_session):
+    with _client_with_registry(tmp_path, monkeypatch, _GRAFANA, _SENTRY) as client:
+        response = client.get("/api/mcp-servers/registry", params={"query": "observability"})
+
+        assert response.status_code == 200
+        grafana, sentry = response.json()
+        assert grafana["name"] == "grafana"
+        assert grafana["registryName"] == "io.github.grafana/mcp-grafana"
+        assert grafana["official"] is True
+        # Declared inputs ride verbatim — the registry owns their shape.
+        assert grafana["headers"] == [_GRAFANA_HEADER]
+        # The url-pinned sentry resolves with the pinned official url.
+        assert sentry["url"] == "https://mcp.sentry.dev/mcp"
+
+
+def test_registry_search_route_maps_unavailability_to_502(tmp_path, monkeypatch, db_session):
+    monkeypatch.setattr(
+        registry, "_http", _client_returning(lambda _r: httpx.Response(503, text="down"))
+    )
+    app = configure_app_for_test(settings=make_settings(tmp_path))
+    with TestClient(app) as client:
+        response = client.get("/api/mcp-servers/registry", params={"query": "grafana"})
+
+        assert response.status_code == 502
+        assert "registry search" in response.json()["detail"]
+
+
+def test_add_from_registry_writes_the_row_and_redacts_the_secret(tmp_path, monkeypatch, db_session):
+    with _client_with_registry(tmp_path, monkeypatch, _ACME_ENTRY) as client:
+        created = client.post(
+            "/api/mcp-servers/registry",
+            json={
+                "name": "observer",
+                "registry": "com.acme/observer",
+                "headers": {"X-Api-Key": "acme-api-secret", "X-Region": "eu"},
+            },
+        )
+
+        assert created.status_code == 200
+        body = created.json()
+        # Header-auth'd: no bearer to miss, ready and enabled immediately.
+        assert body["tokenSource"] == ""
+        assert body["isEnabled"] is True
+        assert body["hasToken"] is True
+        assert "acme-api-secret" not in created.text
+
+        listed = client.get("/api/mcp-servers")
+        assert "acme-api-secret" not in listed.text
+
+    # The row: url from the registry (never the client), values split by the
+    # spec's secrecy — the plain one readable, the secret one ciphertext at
+    # rest and redacted in repr.
+    row = McpServer.get_by_name("observer")
+    assert row.url == "https://mcp.acme.com/mcp"
+    assert row.headers == {"X-Region": "eu"}
+    assert "acme-api-secret" not in repr(row.secret_headers)
+    assert row.secret_headers["X-Api-Key"] == "acme-api-secret"
+
+
+def test_add_from_registry_oauth_candidate_ships_dark_and_connects(
+    tmp_path, monkeypatch, db_session
+):
+    with _client_with_registry(tmp_path, monkeypatch, _GRAFANA) as client:
+        created = client.post(
+            "/api/mcp-servers/registry",
+            json={
+                "name": "grafana",
+                "registry": "io.github.grafana/mcp-grafana",
+                "headers": {"X-Grafana-URL": "https://acme.grafana.net"},
+            },
+        )
+
+        assert created.status_code == 200
+        body = created.json()
+        assert body["tokenSource"] == "oauth"
+        # Dark until its Connect lands — an enabled unconnected oauth server
+        # would fail every delivery.
+        assert body["isEnabled"] is False
+        assert body["hasToken"] is False
+
+        # The added row connects through the existing flow, against the row's
+        # registry-supplied url.
+        begun = []
+
+        async def fake_begin_connect(name, server_url, endpoint):
+            begun.append((name, server_url, endpoint))
+            return "https://consent.example/authorize"
+
+        monkeypatch.setattr("druks.mcp.oauth.begin_connect", fake_begin_connect)
+        connect = client.post("/api/mcp-servers/grafana/connect")
+
+        assert connect.status_code == 200
+        assert connect.json()["authorizationUrl"] == "https://consent.example/authorize"
+        assert begun == [("grafana", "https://mcp.grafana.com/mcp", "http://druks.test")]
+
+    row = McpServer.get_by_name("grafana")
+    assert row.headers == {"X-Grafana-URL": "https://acme.grafana.net"}
+
+
+def test_add_from_registry_rejects_missing_required_and_unknown_headers(
+    tmp_path, monkeypatch, db_session
+):
+    with _client_with_registry(tmp_path, monkeypatch, _ACME_ENTRY) as client:
+        # Required X-Api-Key blank → named 422; nothing persisted.
+        missing = client.post(
+            "/api/mcp-servers/registry",
+            json={
+                "name": "observer",
+                "registry": "com.acme/observer",
+                "headers": {"X-Api-Key": "  ", "X-Region": "eu"},
+            },
+        )
+        assert missing.status_code == 422
+        assert "X-Api-Key" in missing.json()["detail"]
+
+        # An unknown header is a client bug even when its value is blank.
+        for bogus_value in ("v", ""):
+            unknown = client.post(
+                "/api/mcp-servers/registry",
+                json={
+                    "name": "observer",
+                    "registry": "com.acme/observer",
+                    "headers": {"X-Api-Key": "k", "X-Bogus": bogus_value},
+                },
+            )
+            assert unknown.status_code == 422
+            assert "X-Bogus" in unknown.json()["detail"]
+
+        assert not McpServer.get_by_name("observer")
+
+
+def test_add_from_registry_rejects_an_entry_without_an_http_remote(
+    tmp_path, monkeypatch, db_session
+):
+    # stdio/oci-only and unpinned: resolvable in search, not installable.
+    with _client_with_registry(tmp_path, monkeypatch, _entry("com.mcparmory/grafana")) as client:
+        created = client.post(
+            "/api/mcp-servers/registry",
+            json={"name": "grafana", "registry": "com.mcparmory/grafana", "headers": {}},
+        )
+
+        assert created.status_code == 404
+        assert "not installable" in created.json()["detail"]
+
+
+def test_removing_a_connected_row_drops_its_grant(tmp_path, monkeypatch, db_session):
+    with _client_with_registry(tmp_path, monkeypatch, _GRAFANA) as client:
+        client.post(
+            "/api/mcp-servers/registry",
+            json={"name": "grafana", "registry": "io.github.grafana/mcp-grafana", "headers": {}},
+        )
+        McpOauthGrant.store(
+            server_name="grafana",
+            refresh_token="rt",
+            token_endpoint="https://as.example/token",
+            resource="https://mcp.grafana.com/mcp",
+            client_id="cid",
+        )
+
+        assert client.delete("/api/mcp-servers/grafana").status_code == 204
+
+    # An orphan grant would revive as this name's credential on re-add.
+    assert not McpServer.get_by_name("grafana")
+    assert not McpOauthGrant.get_by_server("grafana")
diff --git a/backend/tests/test_mcp_servers.py b/backend/tests/test_mcp_servers.py
index 5b0f8067..a43806b1 100644
--- a/backend/tests/test_mcp_servers.py
+++ b/backend/tests/test_mcp_servers.py
@@ -336,7 +336,7 @@ async def test_bearerless_server_delivers_without_a_bearer(db_session):
 def test_bearerless_server_resolves_ready_with_its_headers(db_session):
     _grafana_shaped_server()
 
-    grafana = next(s for s in McpServer.list_resolved() if s["name"] == "grafana")
+    grafana = McpServer.get_resolved()["grafana"]
     assert grafana["token_source"] == ""
     assert grafana["headers"] == {"X-Grafana-URL": "https://acme.grafana.net"}
     # Nothing blocks delivery auth — the secret header is stored — so the
@@ -483,7 +483,7 @@ def test_packaged_catalog_ships_linear_disabled(registry_state, db_session):
     load_mcp_catalog(PACKAGED_MCP_CATALOG)
 
     assert "github" not in mcp_servers
-    builtins = [s for s in McpServer.list_resolved() if s["builtin"]]
+    builtins = [s for s in McpServer.get_resolved().values() if s["builtin"]]
     assert [s["name"] for s in builtins] == ["linear"]
     linear = builtins[0]
     assert linear["url"] == "https://mcp.linear.app/mcp"
@@ -590,7 +590,7 @@ def test_db_overlay_still_disables_a_catalog_entry(tmp_path, registry_state, db_
 
     McpServer.create(name="figma_test", url="https://mcp.figma.test/", is_enabled=False)
 
-    resolved = next(s for s in McpServer.list_resolved() if s["name"] == "figma_test")
+    resolved = McpServer.get_resolved()["figma_test"]
     assert resolved["builtin"] is True
     assert "figma_test" not in {s["name"] for s in McpServer.list_enabled()}
 
@@ -609,7 +609,7 @@ def test_catalog_enabled_false_ships_the_entry_dark(tmp_path, registry_state, db
         )
     )
 
-    resolved = {s["name"]: s for s in McpServer.list_resolved()}
+    resolved = McpServer.get_resolved()
     assert resolved["dark_test"]["is_enabled"] is False
     assert resolved["lit_test"]["is_enabled"] is True
     enabled_names = {s["name"] for s in McpServer.list_enabled()}
diff --git a/backend/tests/test_secrets.py b/backend/tests/test_secrets.py
index 68165db1..6559bcc2 100644
--- a/backend/tests/test_secrets.py
+++ b/backend/tests/test_secrets.py
@@ -49,7 +49,7 @@ def test_stored_secrets_are_ciphertext_and_reads_restore_them(db_engine, db_sess
     assert row.token.decrypt() == _TOKEN
     # The resolved view every consumer reads carries the Secret itself, so it
     # plaintext exists only where decrypt() is called.
-    resolved = next(s for s in McpServer.list_resolved() if s["name"] == "linear")
+    resolved = McpServer.get_resolved()["linear"]
     assert resolved["token"].decrypt() == _TOKEN
     assert resolved["has_token"] is True