feat(mcp): add index routing to search-records (RAAE-1606)

redisvl/mcp/server.py

@@ -15,6 +15,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
@@ -246,6 +247,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,96 @@
+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.
+    """
+    # Mirror resolve_binding: with no bindings the server is not started (or has
+    # been torn down), so fail loudly rather than return an empty list that a
+    # client could misread as "no indexes configured".
+    if not server._bindings:
+        raise RuntimeError("MCP server has not been started")
+    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."""
+
+    async def list_indexes_tool():
+        """FastMCP wrapper for the `list-indexes` tool."""
+        auth_config = getattr(server, "auth_config", None)
+        read_scope = auth_config.read_scope if auth_config is not None else None
+        ensure_tool_scope(server, read_scope)
+        return list_indexes(server)
+
+    server.tool(name="list-indexes", description=DEFAULT_LIST_INDEXES_DESCRIPTION)(
+        list_indexes_tool
+    )

redisvl/mcp/tools/search.py

@@ -56,12 +56,15 @@ def _build_search_tool_description(
     """Build the `search-records` description from static text plus schema hints.
 
     With multiple bindings configured the schema is ambiguous (the caller picks
-    an index per call via `list-indexes`), so `schema` is None and only the
-    base description is returned.
+    an index per call via `list-indexes`), so per-field hints are omitted and a
+    routing note is appended instead.
     """
     description = (base_description or DEFAULT_SEARCH_DESCRIPTION).strip()
     if schema is None:
-        return description
+        return (
+            description + " Multiple indexes are configured: call list-indexes "
+            "first, then pass the chosen index id as the `index` argument."
+        )
 
     # `exists` is currently accepted for any schema field in the MCP object filter.
     exists_fields = [field.name for field in schema.fields.values()]
@@ -427,14 +430,21 @@ async def search_records(
     server: Any,
     *,
     query: str,
+    index: str | None = None,
     limit: int | None = None,
     offset: int = 0,
     filter: str | dict[str, Any] | None = None,
     return_fields: list[str] | None = None,
 ) -> dict[str, Any]:
-    """Execute `search-records` against the selected Redis index binding."""
+    """Execute `search-records` against the selected Redis index binding.
+
+    ``index`` names the logical binding to query. It is optional when exactly
+    one binding is configured (preserving single-index behavior) and required
+    when multiple bindings exist. The resolved logical id is echoed back in the
+    response so multi-index clients can confirm routing.
+    """
     try:
-        rt = server.resolve_binding(None)
+        rt = server.resolve_binding(index)
         effective_limit, effective_return_fields = _validate_request(
             query=query,
             limit=limit,
@@ -458,6 +468,7 @@ async def search_records(
         )
         sliced_results = raw_results[offset : offset + effective_limit]
         return {
+            "index": rt.binding_id,
             "search_type": search_type,
             "offset": offset,
             "limit": effective_limit,
@@ -485,6 +496,7 @@ def register_search_tool(server: Any, schema: IndexSchema | None) -> None:
 
     async def search_records_tool(
         query: str,
+        index: str | None = None,
         limit: int | None = None,
         offset: int = 0,
         filter: str | dict[str, Any] | None = None,
@@ -497,6 +509,7 @@ async def search_records_tool(
         return await search_records(
             server,
             query=query,
+            index=index,
             limit=limit,
             offset=offset,
             filter=filter,

tests/integration/test_mcp/test_search_tool.py

@@ -214,6 +214,108 @@ async def started(search: dict, **kwargs) -> RedisVLMCPServer:
         await server.shutdown()
 
 
+@pytest.fixture
+async def multi_index_server(
+    monkeypatch, searchable_index, fulltext_only_index, tmp_path, redis_url
+):
+    monkeypatch.setattr(
+        "redisvl.mcp.server.resolve_vectorizer_class",
+        lambda class_name: FakeVectorizer,
+    )
+
+    config = {
+        "server": {"redis_url": redis_url},
+        "indexes": {
+            "knowledge": {
+                "redis_name": searchable_index.schema.index.name,
+                "search": {"type": "vector"},
+                "vectorizer": {
+                    "class": "FakeVectorizer",
+                    "model": "fake-model",
+                    "dims": 3,
+                },
+                "runtime": {
+                    "text_field_name": "content",
+                    "vector_field_name": "embedding",
+                    "default_embed_text_field": "content",
+                    "default_limit": 2,
+                    "max_limit": 5,
+                },
+            },
+            "tickets": {
+                "redis_name": fulltext_only_index.schema.index.name,
+                "search": {"type": "fulltext", "params": {"stopwords": None}},
+                "runtime": {
+                    "text_field_name": "content",
+                    "vector_field_name": None,
+                    "default_embed_text_field": None,
+                    "default_limit": 2,
+                    "max_limit": 5,
+                },
+            },
+        },
+    }
+    config_path = tmp_path / "multi-index-search.yaml"
+    config_path.write_text(yaml.safe_dump(config), encoding="utf-8")
+
+    server = RedisVLMCPServer(MCPSettings(config=str(config_path)))
+    await server.startup()
+    try:
+        yield server
+    finally:
+        await server.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_search_records_routes_to_named_binding(multi_index_server):
+    knowledge = await search_records(
+        multi_index_server,
+        query="science",
+        index="knowledge",
+        return_fields=["content", "category"],
+    )
+    assert knowledge["index"] == "knowledge"
+    assert knowledge["search_type"] == "vector"
+    assert knowledge["results"]
+
+    tickets = await search_records(
+        multi_index_server,
+        query="science",
+        index="tickets",
+        return_fields=["content", "category"],
+    )
+    assert tickets["index"] == "tickets"
+    assert tickets["search_type"] == "fulltext"
+    assert tickets["results"]
+
+
+@pytest.mark.asyncio
+async def test_search_records_requires_index_when_multiple_bindings(multi_index_server):
+    with pytest.raises(RedisVLMCPError) as exc_info:
+        await search_records(multi_index_server, query="science")
+
+    assert exc_info.value.code == MCPErrorCode.INVALID_REQUEST
+
+
+@pytest.mark.asyncio
+async def test_search_records_rejects_unknown_index_on_multi_binding(
+    multi_index_server,
+):
+    with pytest.raises(RedisVLMCPError) as exc_info:
+        await search_records(multi_index_server, query="science", index="missing")
+
+    assert exc_info.value.code == MCPErrorCode.INVALID_REQUEST
+
+
+@pytest.mark.asyncio
+async def test_search_records_single_binding_echoes_index_when_omitted(started_server):
+    server = await started_server({"type": "vector"})
+
+    response = await search_records(server, query="science")
+
+    assert response["index"] == "knowledge"
+
+
 @pytest.mark.asyncio
 async def test_search_records_vector_success_with_pagination_and_projection(
     started_server,

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 (
@@ -730,3 +731,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()