create_index

Instructions

Implementation Reference

• src/foundry_reverse/server.py:277-293 (registration)

  The MCP tool 'create_index' is registered via @mcp.tool decorator. The handler function takes index_name and embed_model, then delegates to kn.create_index().

  @mcp.tool(
      name="create_index",
      description=(
          "Create a named local vector index for RAG (Retrieval-Augmented Generation). "
          "Documents added to this index are embedded via Ollama."
      ),
  )
  def create_index(
      index_name: str,
      embed_model: str = "nomic-embed-text",
  ) -> dict[str, Any]:
      """
      Args:
          index_name: Unique name for the index.
          embed_model: Ollama embedding model to use (must support /api/embeddings).
      """
      return kn.create_index(name=index_name, embed_model=embed_model)

• src/foundry_reverse/knowledge.py:99-104 (handler)

  Core implementation of create_index. Creates a new Index dataclass instance in the _indexes dict (or returns 'already_exists'), then persists to JSON store.

  def create_index(name: str, embed_model: str = DEFAULT_EMBED_MODEL) -> dict[str, Any]:
      if name in _indexes:
          return {"status": "already_exists", "name": name}
      _indexes[name] = Index(name=name, embed_model=embed_model)
      _save()
      return {"status": "created", "name": name, "embed_model": embed_model}

• src/foundry_reverse/knowledge.py:24-36 (schema)

  Dataclass schemas for Index (name, embed_model, documents) and Document (id, text, metadata, embedding) used by create_index and related knowledge operations.

  @dataclass
  class Document:
      id: str
      text: str
      metadata: dict[str, Any]
      embedding: list[float]
  
  
  @dataclass
  class Index:
      name: str
      embed_model: str
      documents: list[Document] = field(default_factory=list)

• src/foundry_reverse/knowledge.py:44-59 (helper)

  _save() helper: persists the in-memory _indexes dict to a JSON file, called by create_index after creating a new index.

  def _save() -> None:
      data = {}
      for idx_name, idx in _indexes.items():
          data[idx_name] = {
              "embed_model": idx.embed_model,
              "documents": [
                  {
                      "id": d.id,
                      "text": d.text,
                      "metadata": d.metadata,
                      "embedding": d.embedding,
                  }
                  for d in idx.documents
              ],
          }
      STORE_PATH.write_text(json.dumps(data))

• src/foundry_reverse/knowledge.py:62-83 (helper)

  _load() helper: loads persisted indexes from JSON file into _indexes dict on module import, ensuring create_index sees existing indexes.

  def _load() -> None:
      if not STORE_PATH.exists():
          return
      data = json.loads(STORE_PATH.read_text())
      for idx_name, idx_data in data.items():
          docs = [
              Document(
                  id=d["id"],
                  text=d["text"],
                  metadata=d["metadata"],
                  embedding=d["embedding"],
              )
              for d in idx_data["documents"]
          ]
          _indexes[idx_name] = Index(
              name=idx_name,
              embed_model=idx_data["embed_model"],
              documents=docs,
          )
  
  
  _load()