diff --git a/pyproject.toml b/pyproject.toml
index 7e35f19e..40731b4e 100644
--- a/pyproject.toml
+++ b/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"]
@@ -148,4 +152,3 @@ filterwarnings = [
 warn_unused_configs = true
 ignore_missing_imports = true
 exclude = ["env", "venv", ".venv"]
-
diff --git a/redisvl/mcp/__init__.py b/redisvl/mcp/__init__.py
new file mode 100644
index 00000000..f86933e6
--- /dev/null
+++ b/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",
+]
diff --git a/redisvl/mcp/config.py b/redisvl/mcp/config.py
new file mode 100644
index 00000000..f01030ba
--- /dev/null
+++ b/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)
diff --git a/redisvl/mcp/errors.py b/redisvl/mcp/errors.py
new file mode 100644
index 00000000..54fb59bc
--- /dev/null
+++ b/redisvl/mcp/errors.py
@@ -0,0 +1,69 @@
+import asyncio
+from enum import Enum
+from typing import Any, Dict, Optional
+
+from pydantic import ValidationError
+from redis.exceptions import RedisError
+
+from redisvl.exceptions import RedisSearchError
+
+
+class MCPErrorCode(str, Enum):
+    """Stable internal error codes exposed by the MCP framework."""
+
+    INVALID_REQUEST = "invalid_request"
+    DEPENDENCY_MISSING = "dependency_missing"
+    BACKEND_UNAVAILABLE = "backend_unavailable"
+    INTERNAL_ERROR = "internal_error"
+
+
+class RedisVLMCPError(Exception):
+    """Framework-facing exception carrying a stable MCP error contract."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        code: MCPErrorCode,
+        retryable: bool,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        super().__init__(message)
+        self.code = code
+        self.retryable = retryable
+        self.metadata = metadata or {}
+
+
+def map_exception(exc: Exception) -> RedisVLMCPError:
+    """Map framework exceptions into deterministic MCP-facing exceptions."""
+    if isinstance(exc, RedisVLMCPError):
+        return exc
+
+    if isinstance(exc, (ValidationError, ValueError, FileNotFoundError)):
+        return RedisVLMCPError(
+            str(exc),
+            code=MCPErrorCode.INVALID_REQUEST,
+            retryable=False,
+        )
+
+    if isinstance(exc, ImportError):
+        return RedisVLMCPError(
+            str(exc),
+            code=MCPErrorCode.DEPENDENCY_MISSING,
+            retryable=False,
+        )
+
+    if isinstance(
+        exc, (TimeoutError, asyncio.TimeoutError, RedisSearchError, RedisError)
+    ):
+        return RedisVLMCPError(
+            str(exc),
+            code=MCPErrorCode.BACKEND_UNAVAILABLE,
+            retryable=True,
+        )
+
+    return RedisVLMCPError(
+        str(exc),
+        code=MCPErrorCode.INTERNAL_ERROR,
+        retryable=False,
+    )
diff --git a/redisvl/mcp/server.py b/redisvl/mcp/server.py
new file mode 100644
index 00000000..e91eed79
--- /dev/null
+++ b/redisvl/mcp/server.py
@@ -0,0 +1,136 @@
+import asyncio
+from importlib import import_module
+from typing import Any, Awaitable, Optional, Type
+
+from redisvl.index import AsyncSearchIndex
+from redisvl.mcp.config import MCPConfig, load_mcp_config
+from redisvl.mcp.settings import MCPSettings
+
+try:
+    from mcp.server.fastmcp import FastMCP
+except ImportError:
+
+    class FastMCP:  # type: ignore[no-redef]
+        """Import-safe stand-in used when the optional MCP SDK is unavailable."""
+
+        def __init__(self, *args, **kwargs):
+            self.args = args
+            self.kwargs = kwargs
+
+
+def resolve_vectorizer_class(class_name: str) -> Type[Any]:
+    """Resolve a vectorizer class from the public RedisVL vectorizer module."""
+    vectorize_module = import_module("redisvl.utils.vectorize")
+    try:
+        return getattr(vectorize_module, class_name)
+    except AttributeError as exc:
+        raise ValueError(f"Unknown vectorizer class: {class_name}") from exc
+
+
+class RedisVLMCPServer(FastMCP):
+    """MCP server exposing RedisVL vector search capabilities.
+
+    This server manages the lifecycle of a Redis vector index and an embedding
+    vectorizer, providing Model Context Protocol (MCP) tools for semantic search
+    operations. It handles configuration loading, connection management,
+    concurrency limits, and graceful shutdown of resources.
+    """
+
+    def __init__(self, settings: MCPSettings):
+        """Create a server shell with lazy config, index, and vectorizer state."""
+        super().__init__("redisvl")
+        self.mcp_settings = settings
+        self.config: Optional[MCPConfig] = None
+        self._index: Optional[AsyncSearchIndex] = None
+        self._vectorizer: Optional[Any] = None
+        self._semaphore: Optional[asyncio.Semaphore] = None
+
+    async def startup(self) -> None:
+        """Load config, validate Redis/index state, and initialize dependencies."""
+        self.config = load_mcp_config(self.mcp_settings.config)
+        self._semaphore = asyncio.Semaphore(self.config.runtime.max_concurrency)
+        self._index = AsyncSearchIndex(
+            schema=self.config.to_index_schema(),
+            redis_url=self.config.redis_url,
+        )
+
+        timeout = self.config.runtime.startup_timeout_seconds
+        index_exists = await asyncio.wait_for(self._index.exists(), timeout=timeout)
+        if not index_exists:
+            if self.config.runtime.index_mode == "validate_only":
+                raise ValueError(
+                    f"Index '{self.config.index.name}' does not exist for validate_only mode"
+                )
+            await asyncio.wait_for(self._index.create(), timeout=timeout)
+
+        # Vectorizer construction may perform provider-specific setup, so keep it
+        # off the event loop and bound it with the same startup timeout.
+        self._vectorizer = await asyncio.wait_for(
+            asyncio.to_thread(self._build_vectorizer),
+            timeout=timeout,
+        )
+        self._validate_vectorizer_dims()
+
+    async def shutdown(self) -> None:
+        """Release owned vectorizer and Redis resources."""
+        if self._vectorizer is not None:
+            aclose = getattr(self._vectorizer, "aclose", None)
+            close = getattr(self._vectorizer, "close", None)
+            if callable(aclose):
+                await aclose()
+            elif callable(close):
+                close()
+            self._vectorizer = None
+
+        if self._index is not None:
+            await self._index.disconnect()
+
+    async def get_index(self) -> AsyncSearchIndex:
+        """Return the initialized async index or fail if startup has not run."""
+        if self._index is None:
+            raise RuntimeError("MCP server has not been started")
+        return self._index
+
+    async def get_vectorizer(self) -> Any:
+        """Return the initialized vectorizer or fail if startup has not run."""
+        if self._vectorizer is None:
+            raise RuntimeError("MCP server has not been started")
+        return self._vectorizer
+
+    async def run_guarded(self, operation_name: str, awaitable: Awaitable[Any]) -> Any:
+        """Run a coroutine under the configured concurrency and timeout limits."""
+        del operation_name
+        if self.config is None or self._semaphore is None:
+            raise RuntimeError("MCP server has not been started")
+
+        # The semaphore centralizes backpressure so future tool handlers do not
+        # each need to reimplement request-limiting behavior.
+        async with self._semaphore:
+            return await asyncio.wait_for(
+                awaitable,
+                timeout=self.config.runtime.request_timeout_seconds,
+            )
+
+    def _build_vectorizer(self) -> Any:
+        """Instantiate the configured vectorizer class from validated config."""
+        if self.config is None:
+            raise RuntimeError("MCP server config not loaded")
+
+        vectorizer_class = resolve_vectorizer_class(self.config.vectorizer.class_name)
+        return vectorizer_class(**self.config.vectorizer.to_init_kwargs())
+
+    def _validate_vectorizer_dims(self) -> None:
+        """Fail startup when vectorizer dimensions disagree with schema dimensions."""
+        if self.config is None or self._vectorizer is None:
+            return
+
+        configured_dims = self.config.vector_field_dims
+        actual_dims = getattr(self._vectorizer, "dims", None)
+        if (
+            configured_dims is not None
+            and actual_dims is not None
+            and configured_dims != actual_dims
+        ):
+            raise ValueError(
+                f"Vectorizer dims {actual_dims} do not match configured vector field dims {configured_dims}"
+            )
diff --git a/redisvl/mcp/settings.py b/redisvl/mcp/settings.py
new file mode 100644
index 00000000..14aca88d
--- /dev/null
+++ b/redisvl/mcp/settings.py
@@ -0,0 +1,41 @@
+from typing import Any, Optional, cast
+
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class MCPSettings(BaseSettings):
+    """Environment-backed settings for bootstrapping the MCP server."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="REDISVL_MCP_",
+        extra="ignore",
+    )
+
+    config: str = Field(..., min_length=1)
+    read_only: bool = False
+    tool_search_description: Optional[str] = None
+    tool_upsert_description: Optional[str] = None
+
+    @classmethod
+    def from_env(
+        cls,
+        *,
+        config: Optional[str] = None,
+        read_only: Optional[bool] = None,
+        tool_search_description: Optional[str] = None,
+        tool_upsert_description: Optional[str] = None,
+    ) -> "MCPSettings":
+        """Build settings from explicit overrides plus `REDISVL_MCP_*` env vars."""
+        overrides: dict[str, object] = {}
+        if config is not None:
+            overrides["config"] = config
+        if read_only is not None:
+            overrides["read_only"] = read_only
+        if tool_search_description is not None:
+            overrides["tool_search_description"] = tool_search_description
+        if tool_upsert_description is not None:
+            overrides["tool_upsert_description"] = tool_upsert_description
+
+        # `BaseSettings` fills any missing fields from the configured env prefix.
+        return cls(**cast(dict[str, Any], overrides))
diff --git a/spec/MCP.md b/spec/MCP.md
new file mode 100644
index 00000000..df49a3f4
--- /dev/null
+++ b/spec/MCP.md
@@ -0,0 +1,642 @@
+---
+name: redisvl-mcp-server-spec
+description: Implementation specification for a RedisVL MCP server with deterministic, agent-friendly contracts for development and testing.
+metadata:
+  status: draft
+  audience: RedisVL maintainers and coding agents
+  objective: Define a deterministic, testable MCP server contract so agents can implement safely without relying on implicit behavior.
+---
+
+# RedisVL MCP Server Specification
+
+## Overview
+
+This specification defines a Model Context Protocol (MCP) server for RedisVL that allows MCP clients to search and upsert data in a Redis index.
+
+The server is designed for stdio transport first and must be runnable via:
+
+```bash
+uvx --from redisvl[mcp] rvl mcp --config /path/to/mcp_config.yaml
+```
+
+### Goals
+
+1. Expose RedisVL search capabilities (`vector`, `fulltext`, `hybrid`) through stable MCP tools.
+2. Support controlled write access via an upsert tool.
+3. Provide deterministic contracts for tool inputs, outputs, and errors.
+4. Align implementation with existing RedisVL architecture and CLI patterns.
+
+### Non-Goals (v1)
+
+1. Multi-index routing in a single server process.
+2. Remote transports (SSE/HTTP).
+3. Delete/count/info tools (future scope).
+
+---
+
+## Compatibility Matrix
+
+These are hard compatibility expectations for v1.
+
+| Component | Requirement | Notes |
+|----------|-------------|-------|
+| Python | `>=3.9.2,<3.15` | Match project constraints |
+| RedisVL | current repo version | Server lives inside this package |
+| redis-py | `>=5.0,<7.2` | Already required by project |
+| MCP SDK | `mcp>=1.9.0` | Provides FastMCP |
+| Redis server | Redis Stack / Redis with Search module | Required for all search modes |
+| Hybrid search | Redis `>=8.4.0` and redis-py `>=7.1.0` runtime capability | If unavailable, `hybrid` returns structured error |
+
+---
+
+## Architecture
+
+### Module Structure
+
+```text
+redisvl/
+├── mcp/
+│   ├── __init__.py
+│   ├── server.py             # RedisVLMCPServer
+│   ├── settings.py           # MCPSettings
+│   ├── config.py             # Config models + loader + validation
+│   ├── errors.py             # MCP error mapping helpers
+│   ├── filters.py            # Filter parser (DSL + raw string handling)
+│   └── tools/
+│       ├── __init__.py
+│       ├── search.py         # redisvl-search
+│       └── upsert.py         # redisvl-upsert
+└── cli/
+    ├── main.py               # Add `mcp` command dispatch
+    └── mcp.py                # MCP command handler class
+```
+
+### Dependency Groups
+
+Add optional extras for explicit install intent.
+
+```toml
+[project.optional-dependencies]
+mcp = [
+  "mcp>=1.9.0",
+  "pydantic-settings>=2.0",
+]
+```
+
+Notes:
+- `fulltext`/`hybrid` use `TextQuery`/`HybridQuery`, which rely on NLTK stopwords when defaults are used. If `nltk` is not installed and stopwords are enabled, server must return a structured dependency error.
+- Provider vectorizer dependencies remain provider-specific (`openai`, `cohere`, `vertexai`, etc.).
+
+---
+
+## Configuration
+
+Configuration is composed from environment + YAML:
+
+1. `MCPSettings` from env/CLI.
+2. YAML file referenced by `config` setting.
+3. Env substitution inside YAML with strict validation.
+
+### Environment Variables
+
+| Variable | Type | Default | Description |
+|----------|------|---------|-------------|
+| `REDISVL_MCP_CONFIG` | str | required | Path to MCP YAML config |
+| `REDISVL_MCP_READ_ONLY` | bool | `false` | If true, do not register upsert tool |
+| `REDISVL_MCP_TOOL_SEARCH_DESCRIPTION` | str | default text | MCP tool description override |
+| `REDISVL_MCP_TOOL_UPSERT_DESCRIPTION` | str | default text | MCP tool description override |
+
+### YAML Schema (Normative)
+
+```yaml
+redis_url: redis://localhost:6379
+
+index:
+  name: my_index
+  prefix: doc
+  storage_type: hash
+
+fields:
+  - name: content
+    type: text
+  - name: category
+    type: tag
+  - name: embedding
+    type: vector
+    attrs:
+      algorithm: hnsw
+      dims: 1536
+      distance_metric: cosine
+      datatype: float32
+
+vectorizer:
+  class: OpenAITextVectorizer
+  model: text-embedding-3-small
+  # kwargs passed to vectorizer constructor
+  # for providers using api_config, pass as nested object:
+  # api_config:
+  #   api_key: ${OPENAI_API_KEY}
+
+runtime:
+  # index lifecycle mode:
+  # validate_only | create_if_missing (default)
+  index_mode: create_if_missing
+
+  # required explicit field mapping for tool behavior
+  text_field_name: content
+  vector_field_name: embedding
+  default_embed_field: content
+
+  # request constraints
+  default_limit: 10
+  max_limit: 100
+  max_upsert_records: 64
+
+  # default overwrite behavior for existing vectors
+  skip_embedding_if_present: true
+
+  # timeouts
+  startup_timeout_seconds: 30
+  request_timeout_seconds: 60
+
+  # server-side concurrency guard
+  max_concurrency: 16
+```
+
+### Env Substitution Rules
+
+Supported patterns in YAML values:
+- `${VAR}`: required variable. Fail startup if unset.
+- `${VAR:-default}`: optional variable with fallback.
+
+Unresolved required vars must fail startup with config error.
+
+### Config Validation Rules
+
+Server startup must fail fast if:
+1. Config file missing/unreadable.
+2. YAML invalid.
+3. `runtime.text_field_name` not in schema.
+4. `runtime.vector_field_name` not in schema or not vector type.
+5. `runtime.default_embed_field` not in schema.
+6. `default_limit <= 0` or `max_limit < default_limit`.
+7. `max_upsert_records <= 0`.
+
+---
+
+## Lifecycle and Resource Management
+
+### Startup Sequence (Normative)
+
+On server startup:
+
+1. Load settings and config.
+2. Build `IndexSchema`.
+3. Create `AsyncSearchIndex` with `redis_url`.
+4. Validate Redis connectivity by performing a lightweight call (`info` or equivalent search operation).
+5. Handle index lifecycle:
+   - `validate_only`: verify index exists; fail if missing.
+   - `create_if_missing`: create index when absent; do not overwrite existing index.
+6. Instantiate vectorizer.
+7. Validate vectorizer dimensions match configured vector field dims when available.
+8. Register tools (omit upsert in read-only mode).
+
+### Shutdown Sequence
+
+On shutdown, disconnect Redis client owned by `AsyncSearchIndex` and release vectorizer resources if applicable.
+
+### Concurrency Guard
+
+Tool executions are bounded by an async semaphore (`runtime.max_concurrency`). Requests exceeding capacity wait, then may timeout according to `request_timeout_seconds`.
+
+---
+
+## Filter Contract (Normative)
+
+`redisvl-search.filter` follows RedisVL convention and accepts either:
+- `string`: raw RedisVL/RediSearch filter string (passed through to query filter).
+- `object`: JSON DSL described below.
+
+### Operators
+
+- Logical: `and`, `or`, `not`
+- Comparison: `eq`, `ne`, `gt`, `gte`, `lt`, `lte`, `in`, `like`
+- Utility: `exists`
+
+### Atomic Expression Shape
+
+```json
+{ "field": "category", "op": "eq", "value": "science" }
+```
+
+### Composite Shape
+
+```json
+{
+  "and": [
+    { "field": "category", "op": "eq", "value": "science" },
+    {
+      "or": [
+        { "field": "rating", "op": "gte", "value": 4.5 },
+        { "field": "is_pinned", "op": "eq", "value": true }
+      ]
+    }
+  ]
+}
+```
+
+### Parsing Rules
+
+1. Unknown `op` fails with `invalid_filter`.
+2. Unknown `field` fails with `invalid_filter`.
+3. Type mismatches fail with `invalid_filter`.
+4. Empty logical arrays fail with `invalid_filter`.
+5. Object DSL parser translates to `redisvl.query.filter.FilterExpression`.
+6. String filter is treated as raw filter expression and passed through.
+
+---
+
+## Tools
+
+## Tool: `redisvl-search`
+
+Search records using vector, full-text, or hybrid query.
+
+### Request Contract
+
+| Parameter | Type | Required | Default | Constraints |
+|----------|------|----------|---------|-------------|
+| `query` | str | yes | - | non-empty |
+| `search_type` | enum | no | `vector` | `vector` \| `fulltext` \| `hybrid` |
+| `limit` | int | no | `runtime.default_limit` | `1..runtime.max_limit` |
+| `offset` | int | no | `0` | `>=0` |
+| `filter` | string \\| object | no | `null` | Raw RedisVL filter string or DSL object |
+| `return_fields` | list[str] | no | all non-vector fields | Unknown fields rejected |
+
+### Response Contract
+
+```json
+{
+  "search_type": "vector",
+  "offset": 0,
+  "limit": 10,
+  "results": [
+    {
+      "id": "doc:123",
+      "score": 0.93,
+      "score_type": "vector_distance_normalized",
+      "record": {
+        "content": "The document text...",
+        "category": "science"
+      }
+    }
+  ]
+}
+```
+
+### Search Semantics
+
+- `vector`: embeds `query` with configured vectorizer, builds `VectorQuery`.
+- `fulltext`: builds `TextQuery`.
+- `hybrid`: embeds `query`, builds `HybridQuery`.
+- `hybrid` must fail with structured capability error if runtime support is unavailable.
+
+### Errors
+
+| Code | Meaning | Retryable |
+|------|---------|-----------|
+| `invalid_request` | bad query params | no |
+| `invalid_filter` | filter parse/type failure | no |
+| `dependency_missing` | missing optional lib/provider SDK | no |
+| `capability_unavailable` | hybrid unsupported in runtime | no |
+| `backend_unavailable` | Redis unavailable/timeout | yes |
+| `internal_error` | unexpected failure | maybe |
+
+---
+
+## Tool: `redisvl-upsert`
+
+Upsert records with automatic embedding.
+
+Not registered when read-only mode is enabled.
+
+### Request Contract
+
+| Parameter | Type | Required | Default | Constraints |
+|----------|------|----------|---------|-------------|
+| `records` | list[object] | yes | - | non-empty and `len(records) <= runtime.max_upsert_records` |
+| `id_field` | str | no | `null` | if set, must exist in every record |
+| `embed_field` | str | no | `runtime.default_embed_field` | must exist in every record |
+| `skip_embedding_if_present` | bool | no | `runtime.skip_embedding_if_present` | if false, always re-embed |
+
+### Response Contract
+
+```json
+{
+  "status": "success",
+  "keys_upserted": 3,
+  "keys": ["doc:abc123", "doc:def456", "doc:ghi789"]
+}
+```
+
+### Upsert Semantics
+
+1. Validate input records before writing.
+2. Resolve `embed_field`.
+3. Respect `skip_embedding_if_present` (default true): only generate embeddings for records missing configured vector field.
+4. Populate configured vector field.
+5. Call `AsyncSearchIndex.load`.
+
+### Error Semantics
+
+- Validation failures return `invalid_request`.
+- Provider errors return `dependency_missing` or `internal_error` with actionable message.
+- Redis write failures return `backend_unavailable`.
+- On write failure, response must include `partial_write_possible: true` (conservative signal).
+
+---
+
+## Server Implementation
+
+### Core Class Contract
+
+```python
+class RedisVLMCPServer(FastMCP):
+    settings: MCPSettings
+    config: MCPConfig
+
+    async def startup(self) -> None: ...
+    async def shutdown(self) -> None: ...
+
+    async def get_index(self) -> AsyncSearchIndex: ...
+    async def get_vectorizer(self): ...
+```
+
+Tool implementations must always call `await server.get_index()` and `await server.get_vectorizer()`; never read uninitialized attributes directly.
+
+### Field Mapping Requirements
+
+Server owns these validated values:
+- `text_field_name`
+- `vector_field_name`
+- `default_embed_field`
+
+No implicit auto-detection is allowed in v1.
+
+---
+
+## CLI Integration
+
+Current RedisVL CLI is command-dispatch based (not argparse subparsers), so MCP integration must follow existing pattern.
+
+### User Commands
+
+```bash
+rvl mcp --config path/to/mcp_config.yaml
+rvl mcp --config path/to/mcp_config.yaml --read-only
+```
+
+### Required CLI Changes
+
+1. Add `mcp` command to usage/help in `redisvl/cli/main.py`.
+2. Add `RedisVlCLI.mcp()` method that dispatches to new `MCP` handler class.
+3. Implement `redisvl/cli/mcp.py` similar to existing command modules.
+4. Gracefully report missing optional deps (`pip install redisvl[mcp]`).
+
+---
+
+## Client Configuration Examples
+
+### Claude Desktop
+
+```json
+{
+  "mcpServers": {
+    "redisvl": {
+      "command": "uvx",
+      "args": ["--from", "redisvl[mcp]", "rvl", "mcp", "--config", "/path/to/mcp_config.yaml"],
+      "env": {
+        "OPENAI_API_KEY": "sk-..."
+      }
+    }
+  }
+}
+```
+
+### Claude Agents SDK (Python)
+
+```python
+from agents import Agent
+from agents.mcp import MCPServerStdio
+
+async def main():
+    async with MCPServerStdio(
+        command="uvx",
+        args=["--from", "redisvl[mcp]", "rvl", "mcp", "--config", "mcp_config.yaml"],
+    ) as server:
+        agent = Agent(
+            name="search-agent",
+            instructions="Search and maintain Redis-backed knowledge.",
+            mcp_servers=[server],
+        )
+```
+
+### Google ADK (Python)
+
+```python
+from google.adk.agents import LlmAgent
+from google.adk.tools.mcp_tool import McpToolset
+from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
+from mcp import StdioServerParameters
+
+root_agent = LlmAgent(
+    model="gemini-2.0-flash",
+    name="redis_search_agent",
+    instruction="Search and maintain Redis-backed knowledge using vector search.",
+    tools=[
+        McpToolset(
+            connection_params=StdioConnectionParams(
+                server_params=StdioServerParameters(
+                    command="uvx",
+                    args=["--from", "redisvl[mcp]", "rvl", "mcp", "--config", "/path/to/mcp_config.yaml"],
+                    env={
+                        "OPENAI_API_KEY": "sk-..."  # Or other vectorizer API key
+                    }
+                ),
+            ),
+            # Optional: filter to specific tools
+            # tool_filter=["redisvl-search"]
+        )
+    ],
+)
+```
+
+### n8n
+
+n8n supports MCP servers via the MCP Server Trigger node. Configure the RedisVL MCP server as an external MCP tool source:
+
+1. **Using SSE transport** (if supported in future versions):
+   ```json
+   {
+     "mcpServers": {
+       "redisvl": {
+         "url": "http://localhost:9000/sse"
+       }
+     }
+   }
+   ```
+
+2. **Using stdio transport** (via n8n's Execute Command node as a workaround):
+   Configure a workflow that spawns the MCP server process:
+   ```bash
+   uvx --from redisvl[mcp] rvl mcp --config /path/to/mcp_config.yaml
+   ```
+
+Note: Full n8n MCP client support depends on n8n's MCP implementation. Refer to [n8n MCP documentation](https://docs.n8n.io/integrations/builtin/core-nodes/n8n-nodes-langchain.mcptrigger/) for current capabilities.
+
+---
+
+## Observability and Security
+
+### Logging
+
+- Use structured logs with operation name, latency, and error code.
+- Never log secrets (API keys, auth headers, full DSNs with credentials).
+- Log config path but not raw config values for sensitive keys.
+
+### Timeouts
+
+- Startup timeout: `runtime.startup_timeout_seconds`
+- Tool request timeout: `runtime.request_timeout_seconds`
+
+### Secret Handling
+
+- Support env-injected secrets via `${VAR}` substitution.
+- Fail fast for required missing vars.
+
+---
+
+## Testing Strategy
+
+## Unit Tests (`tests/unit/test_mcp/`)
+
+- `test_settings.py`
+  - env parsing and overrides
+  - read-only behavior
+- `test_config.py`
+  - YAML validation
+  - env substitution success/failure
+  - field mapping validation
+- `test_filters.py`
+  - DSL parsing, invalid operators, type mismatches
+- `test_errors.py`
+  - internal exception -> MCP error code mapping
+
+## Integration Tests (`tests/integration/test_mcp/`)
+
+- `test_server_startup.py`
+  - startup success path
+  - missing index in `validate_only`
+  - create in `create_if_missing`
+- `test_search_tool.py`
+  - vector/fulltext/hybrid success paths
+  - hybrid capability failure path
+  - pagination and field projection
+  - filter behavior
+- `test_upsert_tool.py`
+  - insert/update success
+  - id_field/embed_field validation failures
+  - read-only mode excludes tool
+
+### Deterministic Verification Commands
+
+```bash
+uv run python -m pytest tests/unit/test_mcp -q
+uv run python -m pytest tests/integration/test_mcp -q
+```
+
+---
+
+## Implementation Plan and DoD
+
+### Phase 1: Framework
+
+Deliverables:
+1. `redisvl/mcp/` scaffolding.
+2. Config/settings models with strict validation.
+3. Startup/shutdown lifecycle.
+4. Error mapping helpers.
+
+DoD:
+1. Server boots successfully with valid config.
+2. Server fails fast with actionable config errors.
+3. Unit tests for config/settings pass.
+
+### Phase 2: Search Tool
+
+Deliverables:
+1. `redisvl-search` request/response contract.
+2. Filter parser (JSON DSL + raw string pass-through).
+3. Capability checks for hybrid support.
+
+DoD:
+1. All search modes tested.
+2. Invalid filter returns `invalid_filter`.
+3. Capability failures are deterministic and non-ambiguous.
+
+### Phase 3: Upsert Tool
+
+Deliverables:
+1. `redisvl-upsert` implementation.
+2. Record pre-validation.
+3. Read-only exclusion.
+
+DoD:
+1. Upsert works end-to-end.
+2. Invalid records fail before writes.
+3. Read-only mode verified.
+
+### Phase 4: CLI and Packaging
+
+Deliverables:
+1. `rvl mcp` command via current CLI pattern.
+2. Optional dependency group updates.
+3. User-facing error messages for missing extras.
+
+DoD:
+1. `uvx --from redisvl[mcp] rvl mcp --config ...` runs successfully.
+2. CLI help includes `mcp` command.
+
+### Phase 5: Documentation
+
+Deliverables:
+1. Config reference and examples.
+2. Client setup examples.
+3. Troubleshooting guide with common errors and fixes.
+
+DoD:
+1. Docs reflect normative contracts in this spec.
+2. Examples are executable and tested.
+
+---
+
+## Risks and Mitigations
+
+1. Runtime mismatch for hybrid search.
+   - Mitigation: explicit capability check + clear error code.
+2. Dependency drift across provider vectorizers.
+   - Mitigation: dependency matrix and startup validation.
+3. Ambiguous filter behavior causing agent retries.
+   - Mitigation: explicit raw-string pass-through semantics and deterministic DSL parser errors.
+4. Hidden partial writes during failures.
+   - Mitigation: conservative `partial_write_possible` signaling.
+5. Security and deployment limitations (v1 scope).
+   - This implementation is designed for local/development usage via stdio transport. It does not include:
+     - Authentication/authorization mechanisms (unlike Redis Agent Memory Server which supports OAuth2/JWT).
+     - Remote transports (SSE/HTTP) that would enable multi-tenant or networked deployments.
+     - Rate limiting or request validation beyond basic input constraints.
+   - Mitigation: Document clearly that v1 is intended for local, single-user scenarios. Users requiring production-grade security should consider the official Redis MCP server or wait for future RedisVL MCP versions that may add remote transport and auth support.
+   - For production deployments requiring authentication, users can:
+     - Deploy behind an authenticating proxy.
+     - Use environment-based secrets for Redis and vectorizer credentials.
+     - Restrict network access to the MCP server process.
diff --git a/tests/integration/test_mcp/test_server_startup.py b/tests/integration/test_mcp/test_server_startup.py
new file mode 100644
index 00000000..34317ab2
--- /dev/null
+++ b/tests/integration/test_mcp/test_server_startup.py
@@ -0,0 +1,159 @@
+from pathlib import Path
+
+import pytest
+
+from redisvl.mcp.server import RedisVLMCPServer
+from redisvl.mcp.settings import MCPSettings
+
+
+class FakeVectorizer:
+    def __init__(self, model: str, dims: int = 3, **kwargs):
+        self.model = model
+        self.dims = dims
+        self.kwargs = kwargs
+
+
+@pytest.fixture
+def mcp_config_path(tmp_path: Path, redis_url: str, worker_id: str):
+    def factory(
+        *, index_name: str, index_mode: str = "create_if_missing", vector_dims: int = 3
+    ):
+        config_path = tmp_path / f"{index_name}.yaml"
+        config_path.write_text(
+            f"""
+redis_url: {redis_url}
+index:
+  name: {index_name}
+  prefix: doc
+  storage_type: hash
+fields:
+  - name: content
+    type: text
+  - name: embedding
+    type: vector
+    attrs:
+      algorithm: flat
+      dims: 3
+      distance_metric: cosine
+      datatype: float32
+vectorizer:
+  class: FakeVectorizer
+  model: fake-model
+  dims: {vector_dims}
+runtime:
+  index_mode: {index_mode}
+  text_field_name: content
+  vector_field_name: embedding
+  default_embed_field: content
+""".strip(),
+            encoding="utf-8",
+        )
+        return str(config_path)
+
+    return factory
+
+
+@pytest.mark.asyncio
+async def test_server_startup_success(monkeypatch, mcp_config_path, worker_id):
+    monkeypatch.setattr(
+        "redisvl.mcp.server.resolve_vectorizer_class",
+        lambda class_name: FakeVectorizer,
+    )
+    settings = MCPSettings(
+        config=mcp_config_path(index_name=f"mcp-startup-{worker_id}")
+    )
+    server = RedisVLMCPServer(settings)
+
+    await server.startup()
+
+    index = await server.get_index()
+    vectorizer = await server.get_vectorizer()
+
+    assert await index.exists() is True
+    assert vectorizer.dims == 3
+
+    await server.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_server_validate_only_missing_index(
+    monkeypatch, mcp_config_path, worker_id
+):
+    monkeypatch.setattr(
+        "redisvl.mcp.server.resolve_vectorizer_class",
+        lambda class_name: FakeVectorizer,
+    )
+    settings = MCPSettings(
+        config=mcp_config_path(
+            index_name=f"mcp-missing-{worker_id}",
+            index_mode="validate_only",
+        )
+    )
+    server = RedisVLMCPServer(settings)
+
+    with pytest.raises(ValueError, match="does not exist"):
+        await server.startup()
+
+
+@pytest.mark.asyncio
+async def test_server_create_if_missing_is_idempotent(
+    monkeypatch, mcp_config_path, worker_id
+):
+    monkeypatch.setattr(
+        "redisvl.mcp.server.resolve_vectorizer_class",
+        lambda class_name: FakeVectorizer,
+    )
+    config_path = mcp_config_path(index_name=f"mcp-idempotent-{worker_id}")
+    first = RedisVLMCPServer(MCPSettings(config=config_path))
+    second = RedisVLMCPServer(MCPSettings(config=config_path))
+
+    await first.startup()
+    await first.shutdown()
+    await second.startup()
+
+    assert await (await second.get_index()).exists() is True
+
+    await second.shutdown()
+
+
+@pytest.mark.asyncio
+async def test_server_fails_fast_on_vector_dimension_mismatch(
+    monkeypatch, mcp_config_path, worker_id
+):
+    monkeypatch.setattr(
+        "redisvl.mcp.server.resolve_vectorizer_class",
+        lambda class_name: FakeVectorizer,
+    )
+    settings = MCPSettings(
+        config=mcp_config_path(
+            index_name=f"mcp-dims-{worker_id}",
+            vector_dims=8,
+        )
+    )
+    server = RedisVLMCPServer(settings)
+
+    with pytest.raises(ValueError, match="Vectorizer dims"):
+        await server.startup()
+
+
+@pytest.mark.asyncio
+async def test_server_shutdown_disconnects_owned_client(
+    monkeypatch, mcp_config_path, worker_id
+):
+    monkeypatch.setattr(
+        "redisvl.mcp.server.resolve_vectorizer_class",
+        lambda class_name: FakeVectorizer,
+    )
+    settings = MCPSettings(
+        config=mcp_config_path(index_name=f"mcp-shutdown-{worker_id}")
+    )
+    server = RedisVLMCPServer(settings)
+
+    await server.startup()
+    index = await server.get_index()
+
+    assert index.client is not None
+
+    await server.shutdown()
+
+    assert index.client is None
diff --git a/tests/unit/test_mcp/conftest.py b/tests/unit/test_mcp/conftest.py
new file mode 100644
index 00000000..f5d7e2bc
--- /dev/null
+++ b/tests/unit/test_mcp/conftest.py
@@ -0,0 +1,8 @@
+import pytest
+
+
+@pytest.fixture(scope="session", autouse=True)
+def redis_container():
+    # Shadow the repo-wide autouse Redis container fixture so MCP unit tests stay
+    # pure-unit and do not require Docker; Redis coverage lives in integration tests.
+    yield None
diff --git a/tests/unit/test_mcp/test_config.py b/tests/unit/test_mcp/test_config.py
new file mode 100644
index 00000000..ad718416
--- /dev/null
+++ b/tests/unit/test_mcp/test_config.py
@@ -0,0 +1,208 @@
+from pathlib import Path
+
+import pytest
+
+from redisvl.mcp.config import MCPConfig, load_mcp_config
+from redisvl.schema import IndexSchema
+
+
+def test_load_mcp_config_file_not_found():
+    with pytest.raises(FileNotFoundError):
+        load_mcp_config("/tmp/does-not-exist.yaml")
+
+
+def test_load_mcp_config_invalid_yaml(tmp_path: Path):
+    config_path = tmp_path / "mcp.yaml"
+    config_path.write_text("redis_url: [", encoding="utf-8")
+
+    with pytest.raises(ValueError, match="Invalid MCP config YAML"):
+        load_mcp_config(str(config_path))
+
+
+def test_load_mcp_config_env_substitution(tmp_path: Path, monkeypatch):
+    config_path = tmp_path / "mcp.yaml"
+    config_path.write_text(
+        """
+redis_url: ${REDIS_URL:-redis://localhost:6379}
+index:
+  name: docs
+  prefix: doc
+  storage_type: hash
+fields:
+  - name: content
+    type: text
+  - name: embedding
+    type: vector
+    attrs:
+      algorithm: flat
+      dims: 3
+      distance_metric: cosine
+      datatype: float32
+vectorizer:
+  class: FakeVectorizer
+  model: test-model
+  api_key: ${OPENAI_API_KEY}
+runtime:
+  text_field_name: content
+  vector_field_name: embedding
+  default_embed_field: content
+""".strip(),
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("OPENAI_API_KEY", "secret")
+
+    config = load_mcp_config(str(config_path))
+
+    assert config.redis_url == "redis://localhost:6379"
+    assert config.vectorizer.class_name == "FakeVectorizer"
+    assert config.vectorizer.model == "test-model"
+    assert config.vectorizer.extra_kwargs == {"api_key": "secret"}
+
+
+def test_load_mcp_config_required_env_missing(tmp_path: Path, monkeypatch):
+    config_path = tmp_path / "mcp.yaml"
+    config_path.write_text(
+        """
+redis_url: redis://localhost:6379
+index:
+  name: docs
+  prefix: doc
+  storage_type: hash
+fields:
+  - name: content
+    type: text
+  - name: embedding
+    type: vector
+    attrs:
+      algorithm: flat
+      dims: 3
+      distance_metric: cosine
+      datatype: float32
+vectorizer:
+  class: FakeVectorizer
+  model: ${VECTOR_MODEL}
+runtime:
+  text_field_name: content
+  vector_field_name: embedding
+  default_embed_field: content
+""".strip(),
+        encoding="utf-8",
+    )
+    monkeypatch.delenv("VECTOR_MODEL", raising=False)
+
+    with pytest.raises(ValueError, match="Missing required environment variable"):
+        load_mcp_config(str(config_path))
+
+
+def test_mcp_config_validates_runtime_mapping():
+    with pytest.raises(ValueError, match="runtime.text_field_name"):
+        MCPConfig.model_validate(
+            {
+                "redis_url": "redis://localhost:6379",
+                "index": {"name": "docs", "prefix": "doc", "storage_type": "hash"},
+                "fields": [
+                    {"name": "content", "type": "text"},
+                    {
+                        "name": "embedding",
+                        "type": "vector",
+                        "attrs": {
+                            "algorithm": "flat",
+                            "dims": 3,
+                            "distance_metric": "cosine",
+                            "datatype": "float32",
+                        },
+                    },
+                ],
+                "vectorizer": {"class": "FakeVectorizer", "model": "test-model"},
+                "runtime": {
+                    "text_field_name": "missing",
+                    "vector_field_name": "embedding",
+                    "default_embed_field": "content",
+                },
+            }
+        )
+
+
+def test_mcp_config_validates_vector_field_type():
+    with pytest.raises(ValueError, match="runtime.vector_field_name"):
+        MCPConfig.model_validate(
+            {
+                "redis_url": "redis://localhost:6379",
+                "index": {"name": "docs", "prefix": "doc", "storage_type": "hash"},
+                "fields": [
+                    {"name": "content", "type": "text"},
+                    {"name": "embedding", "type": "text"},
+                ],
+                "vectorizer": {"class": "FakeVectorizer", "model": "test-model"},
+                "runtime": {
+                    "text_field_name": "content",
+                    "vector_field_name": "embedding",
+                    "default_embed_field": "content",
+                },
+            }
+        )
+
+
+def test_mcp_config_validates_limits():
+    with pytest.raises(ValueError, match="max_limit"):
+        MCPConfig.model_validate(
+            {
+                "redis_url": "redis://localhost:6379",
+                "index": {"name": "docs", "prefix": "doc", "storage_type": "hash"},
+                "fields": [
+                    {"name": "content", "type": "text"},
+                    {
+                        "name": "embedding",
+                        "type": "vector",
+                        "attrs": {
+                            "algorithm": "flat",
+                            "dims": 3,
+                            "distance_metric": "cosine",
+                            "datatype": "float32",
+                        },
+                    },
+                ],
+                "vectorizer": {"class": "FakeVectorizer", "model": "test-model"},
+                "runtime": {
+                    "text_field_name": "content",
+                    "vector_field_name": "embedding",
+                    "default_embed_field": "content",
+                    "default_limit": 10,
+                    "max_limit": 5,
+                },
+            }
+        )
+
+
+def test_mcp_config_to_index_schema():
+    config = MCPConfig.model_validate(
+        {
+            "redis_url": "redis://localhost:6379",
+            "index": {"name": "docs", "prefix": "doc", "storage_type": "hash"},
+            "fields": [
+                {"name": "content", "type": "text"},
+                {
+                    "name": "embedding",
+                    "type": "vector",
+                    "attrs": {
+                        "algorithm": "flat",
+                        "dims": 3,
+                        "distance_metric": "cosine",
+                        "datatype": "float32",
+                    },
+                },
+            ],
+            "vectorizer": {"class": "FakeVectorizer", "model": "test-model"},
+            "runtime": {
+                "text_field_name": "content",
+                "vector_field_name": "embedding",
+                "default_embed_field": "content",
+            },
+        }
+    )
+
+    schema = config.to_index_schema()
+
+    assert isinstance(schema, IndexSchema)
+    assert schema.index.name == "docs"
+    assert schema.field_names == ["content", "embedding"]
diff --git a/tests/unit/test_mcp/test_errors.py b/tests/unit/test_mcp/test_errors.py
new file mode 100644
index 00000000..066e3173
--- /dev/null
+++ b/tests/unit/test_mcp/test_errors.py
@@ -0,0 +1,66 @@
+from pydantic import BaseModel, ValidationError
+from redis.exceptions import ConnectionError as RedisConnectionError
+
+from redisvl.exceptions import RedisSearchError
+from redisvl.mcp.errors import MCPErrorCode, RedisVLMCPError, map_exception
+
+
+class SampleModel(BaseModel):
+    value: int
+
+
+def test_validation_errors_map_to_invalid_request():
+    try:
+        SampleModel.model_validate({"value": "bad"})
+    except ValidationError as exc:
+        mapped = map_exception(exc)
+
+    assert mapped.code == MCPErrorCode.INVALID_REQUEST
+    assert mapped.retryable is False
+
+
+def test_import_error_maps_to_dependency_missing():
+    mapped = map_exception(ImportError("missing package"))
+
+    assert mapped.code == MCPErrorCode.DEPENDENCY_MISSING
+    assert mapped.retryable is False
+
+
+def test_redis_errors_map_to_backend_unavailable():
+    mapped = map_exception(RedisSearchError("redis unavailable"))
+
+    assert mapped.code == MCPErrorCode.BACKEND_UNAVAILABLE
+    assert mapped.retryable is True
+
+
+def test_redis_connection_errors_map_to_backend_unavailable():
+    mapped = map_exception(RedisConnectionError("boom"))
+
+    assert mapped.code == MCPErrorCode.BACKEND_UNAVAILABLE
+    assert mapped.retryable is True
+
+
+def test_timeout_error_maps_to_backend_unavailable():
+    mapped = map_exception(TimeoutError("timed out"))
+
+    assert mapped.code == MCPErrorCode.BACKEND_UNAVAILABLE
+    assert mapped.retryable is True
+
+
+def test_unknown_errors_map_to_internal_error():
+    mapped = map_exception(RuntimeError("unexpected"))
+
+    assert mapped.code == MCPErrorCode.INTERNAL_ERROR
+    assert mapped.retryable is False
+
+
+def test_existing_framework_error_is_preserved():
+    original = RedisVLMCPError(
+        "already mapped",
+        code=MCPErrorCode.INVALID_REQUEST,
+        retryable=False,
+    )
+
+    mapped = map_exception(original)
+
+    assert mapped is original
diff --git a/tests/unit/test_mcp/test_settings.py b/tests/unit/test_mcp/test_settings.py
new file mode 100644
index 00000000..cf4b8800
--- /dev/null
+++ b/tests/unit/test_mcp/test_settings.py
@@ -0,0 +1,45 @@
+from pydantic_settings import BaseSettings
+
+from redisvl.mcp.settings import MCPSettings
+
+
+def test_settings_reads_env_defaults(monkeypatch):
+    monkeypatch.setenv("REDISVL_MCP_CONFIG", "/tmp/mcp.yaml")
+    monkeypatch.setenv("REDISVL_MCP_READ_ONLY", "true")
+    monkeypatch.setenv("REDISVL_MCP_TOOL_SEARCH_DESCRIPTION", "search docs")
+    monkeypatch.setenv("REDISVL_MCP_TOOL_UPSERT_DESCRIPTION", "upsert docs")
+
+    settings = MCPSettings()
+
+    assert settings.config == "/tmp/mcp.yaml"
+    assert settings.read_only is True
+    assert settings.tool_search_description == "search docs"
+    assert settings.tool_upsert_description == "upsert docs"
+
+
+def test_settings_explicit_values_override_env(monkeypatch):
+    monkeypatch.setenv("REDISVL_MCP_CONFIG", "/tmp/from-env.yaml")
+    monkeypatch.setenv("REDISVL_MCP_READ_ONLY", "true")
+
+    settings = MCPSettings.from_env(
+        config="/tmp/from-arg.yaml",
+        read_only=False,
+    )
+
+    assert settings.config == "/tmp/from-arg.yaml"
+    assert settings.read_only is False
+
+
+def test_settings_defaults_optional_descriptions(monkeypatch):
+    monkeypatch.delenv("REDISVL_MCP_TOOL_SEARCH_DESCRIPTION", raising=False)
+    monkeypatch.delenv("REDISVL_MCP_TOOL_UPSERT_DESCRIPTION", raising=False)
+    monkeypatch.setenv("REDISVL_MCP_CONFIG", "/tmp/mcp.yaml")
+
+    settings = MCPSettings.from_env()
+
+    assert settings.tool_search_description is None
+    assert settings.tool_upsert_description is None
+
+
+def test_settings_uses_pydantic_base_settings():
+    assert issubclass(MCPSettings, BaseSettings)
diff --git a/uv.lock b/uv.lock
index 34277f94..be3d9dfb 100644
--- a/uv.lock
+++ b/uv.lock
@@ -2657,6 +2657,31 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/27/1a/1f68f9ba0c207934b35b86a8ca3aad8395a3d6dd7921c0686e23853ff5a9/mccabe-0.7.0-py2.py3-none-any.whl", hash = "sha256:6c2d30ab6be0e4a46919781807b4f0d834ebdd6c6e3dca0bda5a15f863427b6e", size = 7350, upload-time = "2022-01-24T01:14:49.62Z" },
 ]
 
+[[package]]
+name = "mcp"
+version = "1.26.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "anyio", marker = "python_full_version >= '3.10'" },
+    { name = "httpx", marker = "python_full_version >= '3.10'" },
+    { name = "httpx-sse", marker = "python_full_version >= '3.10'" },
+    { name = "jsonschema", marker = "python_full_version >= '3.10'" },
+    { name = "pydantic", marker = "python_full_version >= '3.10'" },
+    { name = "pydantic-settings", version = "2.13.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" },
+    { name = "pyjwt", extra = ["crypto"], marker = "python_full_version >= '3.10'" },
+    { name = "python-multipart", marker = "python_full_version >= '3.10'" },
+    { name = "pywin32", marker = "python_full_version >= '3.10' and sys_platform == 'win32'" },
+    { name = "sse-starlette", marker = "python_full_version >= '3.10'" },
+    { name = "starlette", marker = "python_full_version >= '3.10'" },
+    { name = "typing-extensions", marker = "python_full_version >= '3.10'" },
+    { name = "typing-inspection", marker = "python_full_version >= '3.10'" },
+    { name = "uvicorn", marker = "python_full_version >= '3.10' and sys_platform != 'emscripten'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/fc/6d/62e76bbb8144d6ed86e202b5edd8a4cb631e7c8130f3f4893c3f90262b10/mcp-1.26.0.tar.gz", hash = "sha256:db6e2ef491eecc1a0d93711a76f28dec2e05999f93afd48795da1c1137142c66", size = 608005, upload-time = "2026-01-24T19:40:32.468Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/fd/d9/eaa1f80170d2b7c5ba23f3b59f766f3a0bb41155fbc32a69adfa1adaaef9/mcp-1.26.0-py3-none-any.whl", hash = "sha256:904a21c33c25aa98ddbeb47273033c435e595bbacfdb177f4bd87f6dceebe1ca", size = 233615, upload-time = "2026-01-24T19:40:30.652Z" },
+]
+
 [[package]]
 name = "mdit-py-plugins"
 version = "0.4.2"
@@ -4412,6 +4437,44 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/48/f7/925f65d930802e3ea2eb4d5afa4cb8730c8dc0d2cb89a59dc4ed2fcb2d74/pydantic_core-2.41.4-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:c173ddcd86afd2535e2b695217e82191580663a1d1928239f877f5a1649ef39f", size = 2147775, upload-time = "2025-10-14T10:23:45.406Z" },
 ]
 
+[[package]]
+name = "pydantic-settings"
+version = "2.11.0"
+source = { registry = "https://pypi.org/simple" }
+resolution-markers = [
+    "python_full_version < '3.10'",
+]
+dependencies = [
+    { name = "pydantic", marker = "python_full_version < '3.10'" },
+    { name = "python-dotenv", marker = "python_full_version < '3.10'" },
+    { name = "typing-inspection", marker = "python_full_version < '3.10'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/20/c5/dbbc27b814c71676593d1c3f718e6cd7d4f00652cefa24b75f7aa3efb25e/pydantic_settings-2.11.0.tar.gz", hash = "sha256:d0e87a1c7d33593beb7194adb8470fc426e95ba02af83a0f23474a04c9a08180", size = 188394, upload-time = "2025-09-24T14:19:11.764Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/83/d6/887a1ff844e64aa823fb4905978d882a633cfe295c32eacad582b78a7d8b/pydantic_settings-2.11.0-py3-none-any.whl", hash = "sha256:fe2cea3413b9530d10f3a5875adffb17ada5c1e1bab0b2885546d7310415207c", size = 48608, upload-time = "2025-09-24T14:19:10.015Z" },
+]
+
+[[package]]
+name = "pydantic-settings"
+version = "2.13.1"
+source = { registry = "https://pypi.org/simple" }
+resolution-markers = [
+    "python_full_version >= '3.14'",
+    "python_full_version == '3.13.*'",
+    "python_full_version == '3.12.*'",
+    "python_full_version == '3.11.*'",
+    "python_full_version == '3.10.*'",
+]
+dependencies = [
+    { name = "pydantic", marker = "python_full_version >= '3.10'" },
+    { name = "python-dotenv", marker = "python_full_version >= '3.10'" },
+    { name = "typing-inspection", marker = "python_full_version >= '3.10'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/52/6d/fffca34caecc4a3f97bda81b2098da5e8ab7efc9a66e819074a11955d87e/pydantic_settings-2.13.1.tar.gz", hash = "sha256:b4c11847b15237fb0171e1462bf540e294affb9b86db4d9aa5c01730bdbe4025", size = 223826, upload-time = "2026-02-19T13:45:08.055Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/00/4b/ccc026168948fec4f7555b9164c724cf4125eac006e176541483d2c959be/pydantic_settings-2.13.1-py3-none-any.whl", hash = "sha256:d56fd801823dbeae7f0975e1f8c8e25c258eb75d278ea7abb5d9cebb01b56237", size = 58929, upload-time = "2026-02-19T13:45:06.034Z" },
+]
+
 [[package]]
 name = "pydata-sphinx-theme"
 version = "0.15.4"
@@ -4440,6 +4503,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/c7/21/705964c7812476f378728bdf590ca4b771ec72385c533964653c68e86bdc/pygments-2.19.2-py3-none-any.whl", hash = "sha256:86540386c03d588bb81d44bc3928634ff26449851e99741617ecb9037ee5ec0b", size = 1225217, upload-time = "2025-06-21T13:39:07.939Z" },
 ]
 
+[[package]]
+name = "pyjwt"
+version = "2.11.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/5c/5a/b46fa56bf322901eee5b0454a34343cdbdae202cd421775a8ee4e42fd519/pyjwt-2.11.0.tar.gz", hash = "sha256:35f95c1f0fbe5d5ba6e43f00271c275f7a1a4db1dab27bf708073b75318ea623", size = 98019, upload-time = "2026-01-30T19:59:55.694Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/6f/01/c26ce75ba460d5cd503da9e13b21a33804d38c2165dec7b716d06b13010c/pyjwt-2.11.0-py3-none-any.whl", hash = "sha256:94a6bde30eb5c8e04fee991062b534071fd1439ef58d2adc9ccb823e7bcd0469", size = 28224, upload-time = "2026-01-30T19:59:54.539Z" },
+]
+
+[package.optional-dependencies]
+crypto = [
+    { name = "cryptography", marker = "python_full_version >= '3.10'" },
+]
+
 [[package]]
 name = "pylint"
 version = "3.3.9"
@@ -4531,6 +4608,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/5f/ed/539768cf28c661b5b068d66d96a2f155c4971a5d55684a514c1a0e0dec2f/python_dotenv-1.1.1-py3-none-any.whl", hash = "sha256:31f23644fe2602f88ff55e1f5c79ba497e01224ee7737937930c448e4d0e24dc", size = 20556, upload-time = "2025-06-24T04:21:06.073Z" },
 ]
 
+[[package]]
+name = "python-multipart"
+version = "0.0.22"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/94/01/979e98d542a70714b0cb2b6728ed0b7c46792b695e3eaec3e20711271ca3/python_multipart-0.0.22.tar.gz", hash = "sha256:7340bef99a7e0032613f56dc36027b959fd3b30a787ed62d310e951f7c3a3a58", size = 37612, upload-time = "2026-01-25T10:15:56.219Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/1b/d0/397f9626e711ff749a95d96b7af99b9c566a9bb5129b8e4c10fc4d100304/python_multipart-0.0.22-py3-none-any.whl", hash = "sha256:2b2cd894c83d21bf49d702499531c7bafd057d730c201782048f7945d82de155", size = 24579, upload-time = "2026-01-25T10:15:54.811Z" },
+]
+
 [[package]]
 name = "python-ulid"
 version = "3.1.0"
@@ -4771,7 +4857,7 @@ wheels = [
 
 [[package]]
 name = "redisvl"
-version = "0.14.1"
+version = "0.15.0"
 source = { editable = "." }
 dependencies = [
     { name = "jsonpath-ng" },
@@ -4816,6 +4902,11 @@ cohere = [
 langcache = [
     { name = "langcache" },
 ]
+mcp = [
+    { name = "mcp", marker = "python_full_version >= '3.10'" },
+    { name = "pydantic-settings", version = "2.11.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.10'" },
+    { name = "pydantic-settings", version = "2.13.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" },
+]
 mistralai = [
     { name = "mistralai" },
 ]
@@ -4883,6 +4974,7 @@ requires-dist = [
     { name = "jsonpath-ng", specifier = ">=1.5.0" },
     { name = "langcache", marker = "extra == 'all'", specifier = ">=0.11.0" },
     { name = "langcache", marker = "extra == 'langcache'", specifier = ">=0.11.0" },
+    { name = "mcp", marker = "python_full_version >= '3.10' and extra == 'mcp'", specifier = ">=1.9.0" },
     { name = "mistralai", marker = "extra == 'all'", specifier = ">=1.0.0" },
     { name = "mistralai", marker = "extra == 'mistralai'", specifier = ">=1.0.0" },
     { name = "ml-dtypes", specifier = ">=0.4.0,<1.0.0" },
@@ -4896,6 +4988,7 @@ requires-dist = [
     { name = "protobuf", marker = "extra == 'all'", specifier = ">=5.28.0,<6.0.0" },
     { name = "protobuf", marker = "extra == 'vertexai'", specifier = ">=5.28.0,<6.0.0" },
     { name = "pydantic", specifier = ">=2,<3" },
+    { name = "pydantic-settings", marker = "extra == 'mcp'", specifier = ">=2.0" },
     { name = "python-ulid", specifier = ">=3.0.0" },
     { name = "pyyaml", specifier = ">=5.4,<7.0" },
     { name = "redis", specifier = ">=5.0,<7.2" },
@@ -4909,7 +5002,7 @@ requires-dist = [
     { name = "voyageai", marker = "extra == 'all'", specifier = ">=0.2.2" },
     { name = "voyageai", marker = "extra == 'voyageai'", specifier = ">=0.2.2" },
 ]
-provides-extras = ["mistralai", "openai", "nltk", "cohere", "voyageai", "sentence-transformers", "langcache", "vertexai", "bedrock", "pillow", "sql-redis", "all"]
+provides-extras = ["mcp", "mistralai", "openai", "nltk", "cohere", "voyageai", "sentence-transformers", "langcache", "vertexai", "bedrock", "pillow", "sql-redis", "all"]
 
 [package.metadata.requires-dev]
 dev = [
@@ -6004,6 +6097,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/8f/a6/21b1e19994296ba4a34bc7abaf4fcb40d7e7787477bdfde58cd843594459/sqlglot-28.6.0-py3-none-any.whl", hash = "sha256:8af76e825dc8456a49f8ce049d69bbfcd116655dda3e53051754789e2edf8eba", size = 575186, upload-time = "2026-01-13T17:39:22.327Z" },
 ]
 
+[[package]]
+name = "sse-starlette"
+version = "3.3.2"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "anyio", marker = "python_full_version >= '3.10'" },
+    { name = "starlette", marker = "python_full_version >= '3.10'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/5a/9f/c3695c2d2d4ef70072c3a06992850498b01c6bc9be531950813716b426fa/sse_starlette-3.3.2.tar.gz", hash = "sha256:678fca55a1945c734d8472a6cad186a55ab02840b4f6786f5ee8770970579dcd", size = 32326, upload-time = "2026-02-28T11:24:34.36Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/61/28/8cb142d3fe80c4a2d8af54ca0b003f47ce0ba920974e7990fa6e016402d1/sse_starlette-3.3.2-py3-none-any.whl", hash = "sha256:5c3ea3dad425c601236726af2f27689b74494643f57017cafcb6f8c9acfbb862", size = 14270, upload-time = "2026-02-28T11:24:32.984Z" },
+]
+
 [[package]]
 name = "stack-data"
 version = "0.6.3"
@@ -6018,6 +6124,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/f1/7b/ce1eafaf1a76852e2ec9b22edecf1daa58175c090266e9f6c64afcd81d91/stack_data-0.6.3-py3-none-any.whl", hash = "sha256:d5558e0c25a4cb0853cddad3d77da9891a08cb85dd9f9f91b9f8cd66e511e695", size = 24521, upload-time = "2023-09-30T13:58:03.53Z" },
 ]
 
+[[package]]
+name = "starlette"
+version = "0.52.1"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "anyio", marker = "python_full_version >= '3.10'" },
+    { name = "typing-extensions", marker = "python_full_version >= '3.10' and python_full_version < '3.13'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/c4/68/79977123bb7be889ad680d79a40f339082c1978b5cfcf62c2d8d196873ac/starlette-0.52.1.tar.gz", hash = "sha256:834edd1b0a23167694292e94f597773bc3f89f362be6effee198165a35d62933", size = 2653702, upload-time = "2026-01-18T13:34:11.062Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/81/0d/13d1d239a25cbfb19e740db83143e95c772a1fe10202dda4b76792b114dd/starlette-0.52.1-py3-none-any.whl", hash = "sha256:0029d43eb3d273bc4f83a08720b4912ea4b071087a3b48db01b7c839f7954d74", size = 74272, upload-time = "2026-01-18T13:34:09.188Z" },
+]
+
 [[package]]
 name = "sympy"
 version = "1.14.0"
@@ -6534,6 +6653,20 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/96/94/c31f58c7a7f470d5665935262ebd7455c7e4c7782eb525658d3dbf4b9403/urllib3-2.1.0-py3-none-any.whl", hash = "sha256:55901e917a5896a349ff771be919f8bd99aff50b79fe58fec595eb37bbc56bb3", size = 104579, upload-time = "2023-11-13T12:29:42.719Z" },
 ]
 
+[[package]]
+name = "uvicorn"
+version = "0.41.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "click", version = "8.3.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" },
+    { name = "h11", marker = "python_full_version >= '3.10'" },
+    { name = "typing-extensions", marker = "python_full_version == '3.10.*'" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/32/ce/eeb58ae4ac36fe09e3842eb02e0eb676bf2c53ae062b98f1b2531673efdd/uvicorn-0.41.0.tar.gz", hash = "sha256:09d11cf7008da33113824ee5a1c6422d89fbc2ff476540d69a34c87fab8b571a", size = 82633, upload-time = "2026-02-16T23:07:24.1Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/83/e4/d04a086285c20886c0daad0e026f250869201013d18f81d9ff5eada73a88/uvicorn-0.41.0-py3-none-any.whl", hash = "sha256:29e35b1d2c36a04b9e180d4007ede3bcb32a85fbdfd6c6aeb3f26839de088187", size = 68783, upload-time = "2026-02-16T23:07:22.357Z" },
+]
+
 [[package]]
 name = "virtualenv"
 version = "20.35.3"