feat(mcp): add list-indexes discovery tool (RAAE-1605)

redisvl/mcp/server.py

@@ -14,6 +14,7 @@
 from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError
 from redisvl.mcp.runtime import BindingRuntime
 from redisvl.mcp.settings import MCPSettings
+from redisvl.mcp.tools.list_indexes import register_list_indexes_tool
 from redisvl.mcp.tools.search import register_search_tool
 from redisvl.mcp.tools.upsert import register_upsert_tool
 from redisvl.redis.connection import RedisConnectionFactory, is_version_gte
@@ -248,6 +249,8 @@ def _register_tools(self) -> None:
         if len(self._bindings) == 1:
             search_schema = next(iter(self._bindings.values())).schema
 
+        # Discovery is always available so clients can enumerate indexes.
+        register_list_indexes_tool(self)
         register_search_tool(self, search_schema)
         if not self.mcp_settings.read_only:
             register_upsert_tool(self)

redisvl/mcp/tools/list_indexes.py

@@ -0,0 +1,92 @@
+from typing import TYPE_CHECKING, Any
+
+from redisvl.mcp.auth import ensure_tool_scope
+from redisvl.mcp.runtime import BindingRuntime
+
+if TYPE_CHECKING:
+    from redisvl.mcp.server import RedisVLMCPServer
+
+DEFAULT_LIST_INDEXES_DESCRIPTION = (
+    "List the logical indexes configured on this server. Each entry reports the "
+    "index id, an optional description, whether upsert is available, the "
+    "filterable fields discovered from the index, and any explicitly configured "
+    "limits. Call this first on a multi-index server to choose the correct "
+    "index for search-records or upsert-records."
+)
+
+# Runtime limits surfaced to clients, included only when explicitly configured.
+_LIMIT_FIELDS = ("max_limit", "max_upsert_records")
+
+
+def _binding_fields(binding_runtime: BindingRuntime) -> list[dict[str, str]]:
+    """Return a binding's shared filterable fields from its inspected schema.
+
+    The vector field and the configured default embed-source text field are
+    omitted: they are implementation inputs, not fields a client filters on.
+    """
+    embed_source = binding_runtime.binding.runtime.default_embed_text_field
+    fields: list[dict[str, str]] = []
+    for field in binding_runtime.schema.fields.values():
+        field_type = str(getattr(field.type, "value", field.type))
+        if field_type.lower() == "vector":
+            continue
+        if field.name == embed_source:
+            continue
+        fields.append({"name": field.name, "type": field_type})
+    return fields
+
+
+def _binding_limits(binding_runtime: BindingRuntime) -> dict[str, int]:
+    """Return runtime limits that were explicitly configured for the binding.
+
+    Defaults are intentionally excluded so the output reflects deliberate
+    overrides rather than implementation defaults.
+    """
+    runtime = binding_runtime.binding.runtime
+    configured = runtime.model_fields_set
+    return {
+        name: getattr(runtime, name) for name in _LIMIT_FIELDS if name in configured
+    }
+
+
+def _describe_binding(binding_runtime: BindingRuntime) -> dict[str, Any]:
+    """Build the deterministic discovery payload for a single binding."""
+    entry: dict[str, Any] = {"id": binding_runtime.binding_id}
+    if binding_runtime.binding.description is not None:
+        entry["description"] = binding_runtime.binding.description
+    # Reflects both global read-only and the per-index read_only policy.
+    entry["upsert_available"] = not binding_runtime.effective_read_only
+    entry["fields"] = _binding_fields(binding_runtime)
+    limits = _binding_limits(binding_runtime)
+    if limits:
+        entry["limits"] = limits
+    return entry
+
+
+def list_indexes(server: "RedisVLMCPServer") -> dict[str, Any]:
+    """Return the discovery payload for every configured binding.
+
+    The Redis index name (``redis_name``) is intentionally never exposed.
+    """
+    return {
+        "indexes": [
+            _describe_binding(binding_runtime)
+            for binding_runtime in server._bindings.values()
+        ],
+    }
+
+
+def register_list_indexes_tool(server: "RedisVLMCPServer") -> None:
+    """Register the always-available, read-only `list-indexes` MCP tool."""
+    description = (
+        getattr(server.mcp_settings, "tool_list_indexes_description", None)
+        or DEFAULT_LIST_INDEXES_DESCRIPTION
+    )
+
+    async def list_indexes_tool():
+        """FastMCP wrapper for the `list-indexes` tool."""
+        read_scope = getattr(getattr(server, "auth_config", None), "read_scope", None)
+        ensure_tool_scope(server, read_scope)
+        return list_indexes(server)
+
+    server.tool(name="list-indexes", description=description)(list_indexes_tool)

tests/integration/test_mcp/test_server_startup.py

@@ -9,6 +9,7 @@
 from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError
 from redisvl.mcp.server import RedisVLMCPServer
 from redisvl.mcp.settings import MCPSettings
+from redisvl.mcp.tools.list_indexes import list_indexes
 from redisvl.redis.connection import is_version_gte
 from redisvl.schema import IndexSchema
 from tests.conftest import get_redis_version_async
@@ -723,3 +724,78 @@ async def test_server_startup_fails_when_one_binding_is_invalid(
 
     assert server._lifecycle_state.name == "STOPPED"
     assert server._bindings == {}
+
+
+@pytest.mark.asyncio
+async def test_list_indexes_derives_fields_from_inspected_schema(
+    monkeypatch, existing_index, multi_index_config_path
+):
+    knowledge = await existing_index(index_name="mcp-list-knowledge")
+    tickets = await existing_index(index_name="mcp-list-tickets")
+    monkeypatch.setattr(
+        "redisvl.mcp.server.resolve_vectorizer_class",
+        lambda class_name: FakeVectorizer,
+    )
+    server = RedisVLMCPServer(
+        MCPSettings(
+            config=multi_index_config_path(
+                {
+                    # Vector binding: content is the embed source.
+                    "knowledge": {
+                        "redis_name": knowledge.name,
+                        "description": "Product docs",
+                        "vectorizer": {
+                            "class": "FakeVectorizer",
+                            "model": "fake-model",
+                            "dims": 3,
+                        },
+                        "search": {"type": "vector"},
+                        "runtime": {
+                            "text_field_name": "content",
+                            "vector_field_name": "embedding",
+                            "default_embed_text_field": "content",
+                            "max_limit": 25,
+                        },
+                    },
+                    # Fulltext binding: no embed source, read-only.
+                    "tickets": {
+                        "redis_name": tickets.name,
+                        "read_only": True,
+                        "search": {"type": "fulltext"},
+                        "runtime": {"text_field_name": "content"},
+                    },
+                }
+            )
+        )
+    )
+
+    await server.startup()
+
+    try:
+        result = list_indexes(server)
+        indexes = {entry["id"]: entry for entry in result["indexes"]}
+
+        # Both bindings are discoverable; redis_name is never leaked.
+        assert set(indexes) == {"knowledge", "tickets"}
+        for entry in indexes.values():
+            assert "redis_name" not in entry
+            assert knowledge.name not in entry.values()
+            assert tickets.name not in entry.values()
+
+        # Fields come from the inspected schema. The vector field is always
+        # omitted; the embed-source field is omitted only where configured.
+        knowledge_fields = {f["name"] for f in indexes["knowledge"]["fields"]}
+        tickets_fields = {f["name"] for f in indexes["tickets"]["fields"]}
+        assert "embedding" not in knowledge_fields
+        assert "embedding" not in tickets_fields
+        assert "content" not in knowledge_fields  # embed source omitted
+        assert "content" in tickets_fields  # no embed source configured
+
+        # Per-index write policy and explicit limits are reflected.
+        assert indexes["knowledge"]["upsert_available"] is True
+        assert indexes["tickets"]["upsert_available"] is False
+        assert indexes["knowledge"]["limits"] == {"max_limit": 25}
+        assert "limits" not in indexes["tickets"]
+        assert indexes["knowledge"]["description"] == "Product docs"
+    finally:
+        await server.shutdown()