Introduces local RAG Ingest MCP server (rag-ingest-mcp) on top of unstructured-ingest that pairs with transform-mcp (supports Chroma, Qdrant, pgvector)

CHANGELOG.md

@@ -1,3 +1,9 @@
+## [1.6.29]
+
+### Enhancements
+
+- **feat: add `rag-ingest-mcp`, a local RAG MCP server (`unstructured_ingest/mcp/`).** A FastMCP STDIO server that lands Transform MCP Element JSON into a local vector store and serves matched-space similarity search, with three tools (`load_transform_output`, `search`, `list_collections`) and three backends behind one `VectorStore` contract selected by `URAG_STORE_BACKEND`: Chroma (default, embedded), Qdrant (embedded local mode or a server), and pgvector (writes the Unstructured-compliant schema — `id, type, record_id, element_id, text, embeddings` — fitting rows to a pre-created table or auto-provisioning an equivalent one with a cosine HNSW index, and casting the vector to the column's own `vector`/`halfvec` type). A collection is pinned to one embedding space (provider, model, dimension) on first load; loads with a different model are refused and every query is embedded into the pinned space, so corpus and query vectors are always comparable. Writes are conformed by the corresponding ingest connector stagers with deterministic element ids, so re-loading a source upserts instead of duplicating. Includes per-backend getting-started guides under `unstructured_ingest/mcp/examples/`.
+
 ## [1.6.28]
 
 ### Fixes

test/unit/mcp/test_server.py

@@ -0,0 +1,108 @@
+"""Server-level smoke tests for rag-ingest-mcp, offline.
+
+Drives the FastMCP server in-memory with passthrough (pre-embedded) elements,
+so no embedding provider key or running service is needed. Runs against both
+embedded backends; pgvector's server path is covered by the store conformance
+suite plus an integration environment.
+"""
+
+from __future__ import annotations
+
+import json
+
+import anyio
+import pytest
+
+pytest.importorskip("fastmcp", reason="rag-ingest-mcp requires fastmcp")
+
+from fastmcp import Client  # noqa: E402
+
+from unstructured_ingest.mcp.server import mcp  # noqa: E402
+
+VECTORS = {
+    "cats": [1.0, 0.0, 0.0, 0.0],
+    "dogs": [0.0, 1.0, 0.0, 0.0],
+    "stocks": [0.0, 0.0, 1.0, 0.0],
+}
+
+
+@pytest.fixture(params=["chroma", "qdrant"])
+def backend_env(request, monkeypatch, tmp_path) -> str:
+    pytest.importorskip("chromadb" if request.param == "chroma" else "qdrant_client")
+    monkeypatch.setenv("URAG_STORE_BACKEND", request.param)
+    monkeypatch.setenv("URAG_CHROMA_PATH", str(tmp_path / "chroma"))
+    monkeypatch.setenv("URAG_QDRANT_PATH", str(tmp_path / "qdrant"))
+    monkeypatch.setenv("URAG_EMBED_MODEL", "smoke-model")
+    return request.param
+
+
+def write_elements_file(path, texts_to_vectors=VECTORS) -> str:
+    elements = [
+        {
+            "element_id": f"element-{index}",
+            "text": text,
+            "embeddings": vector,
+            "metadata": {"filename": path.name},
+        }
+        for index, (text, vector) in enumerate(texts_to_vectors.items())
+    ]
+    path.write_text(json.dumps(elements))
+    return str(path)
+
+
+def call(tool: str, arguments: dict) -> dict:
+    async def run():
+        async with Client(mcp) as client:
+            return (await client.call_tool(tool, arguments)).data
+
+    return anyio.run(run)
+
+
+def test_load_search_and_list(backend_env, tmp_path):
+    source = write_elements_file(tmp_path / "doc.json")
+    result = call(
+        "load_transform_output",
+        {"sources": [source], "collection": "smoke", "embed_policy": "passthrough"},
+    )
+    assert result["files"][0]["loaded"] == 3, result
+    assert result["space"]["embed_dim"] == 4
+
+    listed = call("list_collections", {})
+    entry = next(c for c in listed["collections"] if c["collection"] == "smoke")
+    assert entry["model"] == "smoke-model"
+    assert entry["dimension"] == 4
+
+
+def test_directory_source_expands_to_all_json(backend_env, tmp_path):
+    docs = tmp_path / "transform-output"
+    docs.mkdir()
+    write_elements_file(docs / "a.json")
+    write_elements_file(docs / "b.json")
+    result = call(
+        "load_transform_output",
+        {"sources": [str(docs)], "collection": "smokedir", "embed_policy": "passthrough"},
+    )
+    loaded = [f for f in result["files"] if "loaded" in f]
+    assert len(loaded) == 2 and all(f["loaded"] == 3 for f in loaded), result
+
+
+def test_space_guard_reports_per_source_error(backend_env, tmp_path):
+    call(
+        "load_transform_output",
+        {
+            "sources": [write_elements_file(tmp_path / "a.json")],
+            "collection": "smokeguard",
+            "embed_policy": "passthrough",
+        },
+    )
+    five_dim = {text: vector + [0.0] for text, vector in VECTORS.items()}
+    result = call(
+        "load_transform_output",
+        {
+            "sources": [write_elements_file(tmp_path / "b.json", five_dim)],
+            "collection": "smokeguard",
+            "embed_policy": "passthrough",
+            "embed_model": "other-model",
+        },
+    )
+    assert "refusing" in result["files"][0]["error"], result

test/unit/mcp/test_stores.py

@@ -0,0 +1,136 @@
+"""Backend conformance suite for rag-ingest-mcp vector stores.
+
+Every backend must pass the same tests; passing this suite is what "supported
+backend" means. Chroma and Qdrant run embedded (a tmp directory, no service).
+pgvector needs a reachable Postgres with the vector extension available — set
+``URAG_TEST_PG_DSN`` (e.g. a throwaway ``pgvector/pgvector`` container) or its
+cases skip.
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+
+import pytest
+
+from unstructured_ingest.data_types.file_data import FileData, SourceIdentifiers
+from unstructured_ingest.mcp.stores import EmbeddingSpace, SpaceMismatch, VectorStore
+
+# Orthogonal 4-dim vectors: cosine ranking is exact, no embedding model needed.
+TEXTS_TO_VECTORS = {
+    "cats": [1.0, 0.0, 0.0, 0.0],
+    "dogs": [0.0, 1.0, 0.0, 0.0],
+    "stocks": [0.0, 0.0, 1.0, 0.0],
+}
+SPACE = EmbeddingSpace(provider="test", model="conformance-model", dimension=4)
+OTHER_SPACE = EmbeddingSpace(provider="test", model="other-model", dimension=5)
+
+
+def make_elements() -> list[dict]:
+    return [
+        {
+            "element_id": f"element-{index}",
+            "text": text,
+            "embeddings": vector,
+            "metadata": {"filename": "doc.txt"},
+        }
+        for index, (text, vector) in enumerate(TEXTS_TO_VECTORS.items())
+    ]
+
+
+def make_file_data(identifier: str = "doc-1") -> FileData:
+    return FileData(
+        identifier=identifier,
+        connector_type="transform_mcp",
+        source_identifiers=SourceIdentifiers(filename=identifier, fullpath=identifier),
+    )
+
+
+@pytest.fixture(params=["chroma", "qdrant", "pgvector"])
+def store(request, tmp_path) -> VectorStore:
+    if request.param == "chroma":
+        pytest.importorskip("chromadb", reason="chroma backend requires the chroma extra")
+        from unstructured_ingest.mcp.stores.chroma import ChromaStore
+
+        yield ChromaStore(str(tmp_path / "chroma"))
+        return
+    if request.param == "qdrant":
+        pytest.importorskip("qdrant_client", reason="qdrant backend requires the qdrant extra")
+        from unstructured_ingest.mcp.stores.qdrant import QdrantStore
+
+        yield QdrantStore(path=str(tmp_path / "qdrant"))
+        return
+    dsn = os.environ.get("URAG_TEST_PG_DSN")
+    if not dsn:
+        pytest.skip("URAG_TEST_PG_DSN not set; skipping pgvector conformance")
+    pytest.importorskip("psycopg", reason="pgvector backend requires psycopg")
+    from unstructured_ingest.mcp.stores.pgvector import PgvectorStore
+
+    pg_store = PgvectorStore(dsn)
+    yield pg_store
+    # The database outlives the test run, so drop everything this test created.
+    import psycopg
+
+    with psycopg.connect(dsn) as conn:
+        for name in pg_store.collections():
+            if name.startswith("conf_"):
+                conn.execute(f'DROP TABLE IF EXISTS "{name}"')
+                conn.execute("DELETE FROM rag_ingest_spaces WHERE collection = %s", (name,))
+
+
+@pytest.fixture
+def collection() -> str:
+    # Unique per test so a shared backend (pgvector) never sees collisions.
+    return f"conf_{uuid.uuid4().hex[:12]}"
+
+
+def load(store: VectorStore, collection: str, identifier: str = "doc-1") -> int:
+    store.ensure_space(collection, SPACE)
+    return store.write(collection, make_elements(), make_file_data(identifier))
+
+
+def test_load_then_search_ranks_by_cosine(store, collection):
+    assert load(store, collection) == 3
+    matches = store.search(collection, [0.9, 0.1, 0.0, 0.0], k=3)
+    assert [m["text"] for m in matches] == ["cats", "dogs", "stocks"]
+    scores = [m["score"] for m in matches]
+    assert scores == sorted(scores, reverse=True)
+    assert scores[0] > 0.9
+
+
+def test_search_returns_metadata(store, collection):
+    load(store, collection)
+    top = store.search(collection, [1.0, 0.0, 0.0, 0.0], k=1)[0]
+    assert top["metadata"], "flattened element metadata should be returned"
+
+
+def test_space_is_pinned_and_guarded(store, collection):
+    load(store, collection)
+    space = store.space_of(collection)
+    assert (space.model, space.dimension) == (SPACE.model, SPACE.dimension)
+    with pytest.raises(SpaceMismatch):
+        store.ensure_space(collection, OTHER_SPACE)
+
+
+def test_reload_upserts_instead_of_duplicating(store, collection):
+    load(store, collection)
+    load(store, collection)
+    matches = store.search(collection, [1.0, 0.0, 0.0, 0.0], k=10)
+    assert len(matches) == 3
+
+
+def test_unembedded_elements_are_dropped(store, collection):
+    store.ensure_space(collection, SPACE)
+    elements = make_elements() + [{"element_id": "bare", "text": "no vector"}]
+    assert store.write(collection, elements, make_file_data()) == 3
+
+
+def test_collections_lists_what_was_loaded(store, collection):
+    load(store, collection)
+    assert collection in store.collections()
+
+
+def test_space_of_unknown_collection_raises(store):
+    with pytest.raises(Exception):
+        store.space_of("conf_never_loaded")

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "1.6.28"  # pragma: no cover
+__version__ = "1.6.29"  # pragma: no cover

unstructured_ingest/mcp/.env.example

@@ -0,0 +1,27 @@
+# --- required for embedding (ingest-side embed and every query) ---
+OPENAI_API_KEY=sk-...
+
+# --- local vector store ---
+# Backend: chroma (default, embedded) | qdrant (embedded or server) | pgvector
+# URAG_STORE_BACKEND=chroma
+
+# chroma: directory where the collections persist. This IS your local RAG corpus.
+URAG_CHROMA_PATH=~/.unstructured-rag/chroma
+
+# qdrant: embedded local mode by default; set URAG_QDRANT_URL to use a server.
+# URAG_QDRANT_PATH=~/.unstructured-rag/qdrant
+# URAG_QDRANT_URL=
+# URAG_QDRANT_API_KEY=
+
+# pgvector: required when URAG_STORE_BACKEND=pgvector.
+# URAG_PG_DSN=postgresql://user:pass@localhost:5432/db
+
+# --- embedding defaults (must match what Transform used, in passthrough mode) ---
+URAG_EMBED_PROVIDER=openai
+URAG_EMBED_MODEL=text-embedding-3-small
+
+# --- optional knobs (defaults shown) ---
+# URAG_STORE_BACKEND=chroma
+# URAG_EMBED_BATCH_SIZE=64
+# URAG_MAX_FETCH_BYTES=536870912
+# OPENAI_BASE_URL=