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

Add an optional `index` argument to the search-records tool so a single
multi-binding MCP server can target a specific logical index. The argument
is optional when exactly one binding is configured (preserving single-index
behavior) and resolves through the same resolve_binding routing used
elsewhere, so an omitted index on a multi-binding server and unknown ids both
surface as invalid_request. The resolved logical id is echoed back as the
`index` field in the response.

- Expose `index` on the FastMCP wrapper param list.
- Append a routing note to the tool description when the schema is ambiguous
  (multiple bindings) directing clients to call list-indexes first.
- Add unit + integration coverage for routing, omitted-index rejection,
  unknown ids, and single-binding backward compatibility.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: vishal-bala

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/unit/test_mcp/test_search_tool_unit.py

@@ -111,8 +111,16 @@ def __init__(
         self.vectorizer = FakeVectorizer() if include_vectorizer else None
         self.registered_tools = []
         self.native_hybrid_supported = False
+        self.resolved_index_ids: list[str | None] = []
 
     def resolve_binding(self, index_id=None):
+        self.resolved_index_ids.append(index_id)
+        if index_id is not None and index_id != "knowledge":
+            raise RedisVLMCPError(
+                f"Unknown index '{index_id}'; available: knowledge",
+                code=MCPErrorCode.INVALID_REQUEST,
+                retryable=False,
+            )
         return BindingRuntime(
             binding_id="knowledge",
             binding=self.config.indexes["knowledge"],
@@ -313,6 +321,7 @@ async def fake_query(query):
     assert built_queries[0]["normalize_vector_distance"] is False
     assert built_queries[0]["ef_runtime"] == 42
     assert response == {
+        "index": "knowledge",
         "search_type": "vector",
         "offset": 0,
         "limit": 2,
@@ -759,6 +768,64 @@ def test_build_search_tool_description_preserves_schema_order_and_excludes_vecto
     assert "embedding" not in description.split("Allowed return_fields: ", 1)[1]
 
 
+@pytest.mark.asyncio
+async def test_search_records_defaults_to_sole_binding_when_index_omitted(monkeypatch):
+    server = FakeServer()
+
+    async def fake_query(query):
+        return []
+
+    server.index.query = fake_query
+
+    response = await search_records(server, query="science")
+
+    assert server.resolved_index_ids == [None]
+    assert response["index"] == "knowledge"
+
+
+@pytest.mark.asyncio
+async def test_search_records_routes_to_named_index(monkeypatch):
+    server = FakeServer()
+
+    async def fake_query(query):
+        return []
+
+    server.index.query = fake_query
+
+    response = await search_records(server, query="science", index="knowledge")
+
+    assert server.resolved_index_ids == ["knowledge"]
+    assert response["index"] == "knowledge"
+
+
+@pytest.mark.asyncio
+async def test_search_records_rejects_unknown_index():
+    server = FakeServer()
+
+    with pytest.raises(RedisVLMCPError) as exc_info:
+        await search_records(server, query="science", index="missing")
+
+    assert exc_info.value.code == MCPErrorCode.INVALID_REQUEST
+    assert server.resolved_index_ids == ["missing"]
+
+
+def test_register_search_tool_wrapper_exposes_index_param():
+    server = FakeServer()
+    register_search_tool(server, server.index.schema)
+
+    annotations = server.registered_tools[0]["fn"].__annotations__
+    assert "index" in annotations
+
+
+def test_build_search_tool_description_appends_routing_note_when_schema_is_ambiguous():
+    description = _build_search_tool_description(None)
+
+    assert "list-indexes" in description
+    assert "`index`" in description
+    # Per-field hints are omitted because the index is ambiguous.
+    assert "Object filter fields" not in description
+
+
 def test_build_search_tool_description_distinguishes_typed_and_exists_support():
     schema = IndexSchema.from_dict(
         {