feat(mcp): add index routing + per-index write policy to upsert-records (RAAE-1607)

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,8 +247,14 @@ 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:
+        # Expose upsert only when at least one binding is writable. A binding is
+        # read-only under global read-only mode or its own read_only policy, both
+        # of which are folded into effective_read_only; the per-call write check
+        # in the tool then rejects writes to any individual read-only binding.
+        if any(not rt.effective_read_only for rt in self._bindings.values()):
             register_upsert_tool(self)
         self._tools_registered = True
 

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,

redisvl/mcp/tools/upsert.py

@@ -249,19 +249,27 @@ async def upsert_records(
     server: Any,
     *,
     records: list[dict[str, Any]],
+    index: str | None = None,
     id_field: str | None = None,
     skip_embedding_if_present: bool | None = None,
 ) -> dict[str, Any]:
-    """Execute `upsert-records` against the selected Redis index binding."""
+    """Execute `upsert-records` against the selected Redis index binding.
+
+    ``index`` names the logical binding to write to. It is optional when exactly
+    one binding is configured and required when multiple exist. Writes to a
+    read-only binding (whether from global read-only mode or the binding's own
+    ``read_only`` policy) are rejected with ``invalid_request``. The resolved
+    logical id is echoed back in the response.
+    """
     try:
-        rt = server.resolve_binding(None)
+        rt = server.resolve_binding(index)
         if rt.effective_read_only:
             raise RedisVLMCPError(
-                "upsert-records is not permitted: binding is read-only",
+                f"index '{rt.binding_id}' is read-only",
                 code=MCPErrorCode.FORBIDDEN,
                 retryable=False,
             )
-        index = rt.index
+        index_obj = rt.index
         runtime = rt.binding.runtime
         effective_skip_embedding = _validate_request(
             runtime=runtime,
@@ -275,7 +283,7 @@ async def upsert_records(
         for record in prepared_records:
             _validate_record(
                 record,
-                index=index,
+                index=index_obj,
                 vector_field_name=runtime.vector_field_name,
             )
         if rt.binding.supports_server_side_embedding:
@@ -332,7 +340,7 @@ async def upsert_records(
         try:
             keys = await server.run_guarded(
                 "upsert-records",
-                index.load(loadable_records, id_field=id_field),
+                index_obj.load(loadable_records, id_field=id_field),
                 timeout_seconds=runtime.request_timeout_seconds,
             )
         except Exception as exc:
@@ -341,6 +349,7 @@ async def upsert_records(
             raise mapped from exc
 
         return {
+            "index": rt.binding_id,
             "status": "success",
             "keys_upserted": len(keys),
             "keys": keys,
@@ -359,6 +368,7 @@ def register_upsert_tool(server: Any) -> None:
 
     async def upsert_records_tool(
         records: list[dict[str, Any]],
+        index: str | None = None,
         id_field: str | None = None,
         skip_embedding_if_present: bool | None = None,
     ):
@@ -369,6 +379,7 @@ async def upsert_records_tool(
         return await upsert_records(
             server,
             records=records,
+            index=index,
             id_field=id_field,
             skip_embedding_if_present=skip_embedding_if_present,
         )

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,