feat(MCP): Implement the base framework for a RedisVL MCP server

docs/api/index.md

@@ -23,5 +23,6 @@ reranker
 cache
 message_history
 router
+cli
 ```
 

docs/user_guide/cli.ipynb

@@ -12,7 +12,9 @@
     "\n",
     "Before running this notebook, be sure to\n",
     "1. Have installed ``redisvl`` and have that environment active for this notebook.\n",
-    "2. Have a running Redis instance with Redis Search enabled"
+    "2. Have a running Redis instance with Redis Search enabled\n",
+    "\n",
+    "For complete command syntax and options, see the [CLI Reference](../api/cli.rst)."
    ]
   },
   {

docs/user_guide/how_to_guides/index.md

@@ -36,6 +36,11 @@ How-to guides are **task-oriented** recipes that help you accomplish specific go
 - [Choose a Storage Type](../05_hash_vs_json.ipynb) -- Hash vs JSON formats and nested data
 :::
 
+:::{grid-item-card} 💻 CLI Operations
+
+- [Manage Indices with the CLI](../cli.ipynb) -- create, inspect, and delete indices from your terminal
+:::
+
 ::::
 
 ## Quick Reference
@@ -53,6 +58,7 @@ How-to guides are **task-oriented** recipes that help you accomplish specific go
 | Improve search accuracy | [Rerank Search Results](../06_rerankers.ipynb) |
 | Optimize index performance | [Optimize Indexes with SVS-VAMANA](../09_svs_vamana.ipynb) |
 | Decide on storage format | [Choose a Storage Type](../05_hash_vs_json.ipynb) |
+| Manage indices from terminal | [Manage Indices with the CLI](../cli.ipynb) |
 
 ```{toctree}
 :hidden:

pyproject.toml

@@ -35,6 +35,10 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
+mcp = [
+    "mcp>=1.9.0 ; python_version >= '3.10'",
+    "pydantic-settings>=2.0",
+]
 mistralai = ["mistralai>=1.0.0"]
 openai = ["openai>=1.1.0"]
 nltk = ["nltk>=3.8.1,<4"]

redisvl/cli/index.py

@@ -30,13 +30,6 @@ def __init__(self):
         parser = argparse.ArgumentParser(usage=self.usage)
 
         parser.add_argument("command", help="Subcommand to run")
-        parser.add_argument(
-            "-f",
-            "--format",
-            help="Output format for info command",
-            type=str,
-            default="rounded_outline",
-        )
         parser = add_index_parsing_options(parser)
 
         args = parser.parse_args(sys.argv[2:])

redisvl/cli/stats.py

@@ -6,7 +6,6 @@
 from redisvl.index import SearchIndex
 from redisvl.schema.schema import IndexSchema
 from redisvl.utils.log import get_logger
-from redisvl.utils.utils import lazy_import
 
 logger = get_logger("[RedisVL]")
 
@@ -42,10 +41,6 @@ class Stats:
 
     def __init__(self):
         parser = argparse.ArgumentParser(usage=self.usage)
-
-        parser.add_argument(
-            "-f", "--format", help="Output format", type=str, default="rounded_outline"
-        )
         parser = add_index_parsing_options(parser)
         args = parser.parse_args(sys.argv[2:])
         try:
@@ -61,7 +56,7 @@ def stats(self, args: Namespace):
             rvl stats -i <index_name> | -s <schema_path>
         """
         index = self._connect_to_index(args)
-        _display_stats(index.info(), output_format=args.format)
+        _display_stats(index.info())
 
     def _connect_to_index(self, args: Namespace) -> SearchIndex:
         # connect to redis
@@ -85,7 +80,7 @@ def _connect_to_index(self, args: Namespace) -> SearchIndex:
         return index
 
 
-def _display_stats(index_info, output_format="rounded_outline"):
+def _display_stats(index_info):
     # Extracting the statistics
     stats_data = [(key, str(index_info.get(key))) for key in STATS_KEYS]
 

redisvl/extensions/cache/base.py

@@ -5,13 +5,13 @@
 """
 
 from collections.abc import Mapping
-from typing import Any, Dict, Optional, Union
+from typing import Any, Dict, Optional
 
 from redis import Redis  # For backwards compatibility in type checking
 from redis.cluster import RedisCluster
 
 from redisvl.redis.connection import RedisConnectionFactory
-from redisvl.types import AsyncRedisClient, SyncRedisClient, SyncRedisCluster
+from redisvl.types import AsyncRedisClient, SyncRedisClient
 
 
 class BaseCache:

redisvl/extensions/router/semantic.py

@@ -1,5 +1,5 @@
 from pathlib import Path
-from typing import Any, Dict, List, Mapping, Optional, Type, Union
+from typing import Any, Dict, List, Optional, Type, Union
 
 import redis.commands.search.reducers as reducers
 import yaml

redisvl/index/index.py

@@ -286,9 +286,13 @@ def _validate_query(self, query: BaseQuery) -> None:
     def _validate_hybrid_query(self, query: Any) -> None:
         """Validate that a hybrid query can be executed."""
         try:
-            from redis.commands.search.hybrid_result import HybridResult
+            from redis.commands.search.hybrid_result import (  # noqa: F401
+                HybridResult as _HybridResult,
+            )
 
             from redisvl.query.hybrid import HybridQuery
+
+            del _HybridResult  # Only imported to check availability
         except (ImportError, ModuleNotFoundError):
             raise ImportError(_HYBRID_SEARCH_ERROR_MESSAGE)
 
@@ -894,14 +898,14 @@ def load(
                 batch_size=batch_size,
                 validate=self._validate_on_load,
             )
-        except SchemaValidationError as e:
+        except SchemaValidationError:
             # Log the detailed validation error with actionable information
             logger.error("Data validation failed during load operation")
             raise
-        except Exception as e:
+        except Exception as exc:
             # Wrap other errors as general RedisVL errors
             logger.exception("Error while loading data to Redis")
-            raise RedisVLError(f"Failed to load data: {str(e)}") from e
+            raise RedisVLError(f"Failed to load data: {str(exc)}") from exc
 
     def fetch(self, id: str) -> Optional[Dict[str, Any]]:
         """Fetch an object from Redis by id.
@@ -1840,14 +1844,14 @@ def add_field(d):
                 batch_size=batch_size,
                 validate=self._validate_on_load,
             )
-        except SchemaValidationError as e:
+        except SchemaValidationError:
             # Log the detailed validation error with actionable information
             logger.error("Data validation failed during load operation")
             raise
-        except Exception as e:
+        except Exception as exc:
             # Wrap other errors as general RedisVL errors
             logger.exception("Error while loading data to Redis")
-            raise RedisVLError(f"Failed to load data: {str(e)}") from e
+            raise RedisVLError(f"Failed to load data: {str(exc)}") from exc
 
     async def fetch(self, id: str) -> Optional[Dict[str, Any]]:
         """Asynchronously etch an object from Redis by id. The id is typically

redisvl/index/storage.py

@@ -1,16 +1,5 @@
 from collections.abc import Collection
-from typing import (
-    Any,
-    Awaitable,
-    Callable,
-    Dict,
-    Iterable,
-    List,
-    Optional,
-    Tuple,
-    Union,
-    cast,
-)
+from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Union
 
 from pydantic import BaseModel, ValidationError
 from redis import __version__ as redis_version

redisvl/mcp/__init__.py

@@ -0,0 +1,14 @@
+from redisvl.mcp.config import MCPConfig, load_mcp_config
+from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError, map_exception
+from redisvl.mcp.server import RedisVLMCPServer
+from redisvl.mcp.settings import MCPSettings
+
+__all__ = [
+    "MCPConfig",
+    "MCPErrorCode",
+    "MCPSettings",
+    "RedisVLMCPError",
+    "RedisVLMCPServer",
+    "load_mcp_config",
+    "map_exception",
+]

redisvl/mcp/config.py

@@ -0,0 +1,168 @@
+import os
+import re
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Union
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+from redisvl.schema.fields import BaseField
+from redisvl.schema.schema import IndexInfo, IndexSchema
+
+_ENV_PATTERN = re.compile(r"\$\{([^}:]+)(?::-([^}]*))?\}")
+
+
+class MCPRuntimeConfig(BaseModel):
+    """Runtime limits and validated field mappings for MCP requests."""
+
+    index_mode: str = "create_if_missing"
+    text_field_name: str
+    vector_field_name: str
+    default_embed_field: str
+    default_limit: int = 10
+    max_limit: int = 100
+    max_upsert_records: int = 64
+    skip_embedding_if_present: bool = True
+    startup_timeout_seconds: int = 30
+    request_timeout_seconds: int = 60
+    max_concurrency: int = 16
+
+    @model_validator(mode="after")
+    def _validate_limits(self) -> "MCPRuntimeConfig":
+        if self.index_mode not in {"validate_only", "create_if_missing"}:
+            raise ValueError(
+                "runtime.index_mode must be validate_only or create_if_missing"
+            )
+        if self.default_limit <= 0:
+            raise ValueError("runtime.default_limit must be greater than 0")
+        if self.max_limit < self.default_limit:
+            raise ValueError(
+                "runtime.max_limit must be greater than or equal to runtime.default_limit"
+            )
+        if self.max_upsert_records <= 0:
+            raise ValueError("runtime.max_upsert_records must be greater than 0")
+        if self.startup_timeout_seconds <= 0:
+            raise ValueError("runtime.startup_timeout_seconds must be greater than 0")
+        if self.request_timeout_seconds <= 0:
+            raise ValueError("runtime.request_timeout_seconds must be greater than 0")
+        if self.max_concurrency <= 0:
+            raise ValueError("runtime.max_concurrency must be greater than 0")
+        return self
+
+
+class MCPVectorizerConfig(BaseModel):
+    """Vectorizer constructor contract loaded from YAML."""
+
+    model_config = ConfigDict(populate_by_name=True, extra="allow")
+
+    class_name: str = Field(alias="class", min_length=1)
+    model: str = Field(..., min_length=1)
+
+    @property
+    def extra_kwargs(self) -> Dict[str, Any]:
+        """Return vectorizer kwargs other than the normalized `class` and `model`."""
+        return dict(self.model_extra or {})
+
+    def to_init_kwargs(self) -> Dict[str, Any]:
+        """Build kwargs suitable for directly instantiating the vectorizer."""
+        return {"model": self.model, **self.extra_kwargs}
+
+
+class MCPConfig(BaseModel):
+    """Validated MCP server configuration loaded from YAML."""
+
+    redis_url: str = Field(..., min_length=1)
+    index: IndexInfo
+    fields: Union[List[Dict[str, Any]], Dict[str, Dict[str, Any]]]
+    vectorizer: MCPVectorizerConfig
+    runtime: MCPRuntimeConfig
+
+    @model_validator(mode="after")
+    def _validate_runtime_mapping(self) -> "MCPConfig":
+        """Ensure runtime field mappings point at explicit schema fields."""
+        schema = self.to_index_schema()
+        field_names = set(schema.field_names)
+
+        if self.runtime.text_field_name not in field_names:
+            raise ValueError(
+                f"runtime.text_field_name '{self.runtime.text_field_name}' not found in schema"
+            )
+
+        if self.runtime.default_embed_field not in field_names:
+            raise ValueError(
+                f"runtime.default_embed_field '{self.runtime.default_embed_field}' not found in schema"
+            )
+
+        vector_field = schema.fields.get(self.runtime.vector_field_name)
+        if vector_field is None:
+            raise ValueError(
+                f"runtime.vector_field_name '{self.runtime.vector_field_name}' not found in schema"
+            )
+        if vector_field.type != "vector":
+            raise ValueError(
+                f"runtime.vector_field_name '{self.runtime.vector_field_name}' must reference a vector field"
+            )
+
+        return self
+
+    def to_index_schema(self) -> IndexSchema:
+        """Convert the MCP config schema fragment into a reusable `IndexSchema`."""
+        return IndexSchema.model_validate(
+            {
+                "index": self.index.model_dump(mode="python"),
+                "fields": self.fields,
+            }
+        )
+
+    @property
+    def vector_field(self) -> BaseField:
+        """Return the configured vector field from the generated index schema."""
+        return self.to_index_schema().fields[self.runtime.vector_field_name]
+
+    @property
+    def vector_field_dims(self) -> Optional[int]:
+        """Return the configured vector dimension when the field exposes one."""
+        attrs = self.vector_field.attrs
+        return getattr(attrs, "dims", None)
+
+
+def _substitute_env(value: Any) -> Any:
+    """Recursively resolve `${VAR}` and `${VAR:-default}` placeholders."""
+    if isinstance(value, dict):
+        return {key: _substitute_env(item) for key, item in value.items()}
+    if isinstance(value, list):
+        return [_substitute_env(item) for item in value]
+    if not isinstance(value, str):
+        return value
+
+    def replace(match: re.Match[str]) -> str:
+        name = match.group(1)
+        default = match.group(2)
+        env_value = os.environ.get(name)
+        if env_value is not None:
+            return env_value
+        if default is not None:
+            return default
+        # Fail fast here so startup never proceeds with partially-resolved config.
+        raise ValueError(f"Missing required environment variable: {name}")
+
+    return _ENV_PATTERN.sub(replace, value)
+
+
+def load_mcp_config(path: str) -> MCPConfig:
+    """Load, substitute, and validate the MCP YAML configuration file."""
+    config_path = Path(path).expanduser()
+    if not config_path.exists():
+        raise FileNotFoundError(f"MCP config file {path} does not exist")
+
+    try:
+        with config_path.open("r", encoding="utf-8") as file:
+            raw_data = yaml.safe_load(file)
+    except yaml.YAMLError as exc:
+        raise ValueError(f"Invalid MCP config YAML: {exc}") from exc
+
+    if not isinstance(raw_data, dict):
+        raise ValueError("Invalid MCP config YAML: root document must be a mapping")
+
+    substituted = _substitute_env(raw_data)
+    return MCPConfig.model_validate(substituted)