index_now

Instructions

    Args:
        path: Target path.  When ``None``, all :attr:`AppContext.config.sources`
            are scanned.
        force: Skip hash comparison and always re-index.
    

Implementation Reference

• src/memorymesh/server/tools/index_now.py:29-108 (handler)

  The core handler function for the index_now tool. Accepts optional path (file or directory) and force flag. If path is given, indexes that file/directory; otherwise iterates all configured sources. Returns n_files_processed, n_chunks, duration_ms, and any errors.

  def index_now(
      path: Annotated[
          str | None,
          "Absolute path to a file or directory to index. "
          "Omit to re-index all configured sources.",
      ] = None,
      force: Annotated[
          bool,
          "Re-index even if the file hash has not changed.",
      ] = False,
  ) -> dict:
      """Trigger an immediate (re-)index of a file, directory, or all sources.
  
      Args:
          path: Target path.  When ``None``, all :attr:`AppContext.config.sources`
              are scanned.
          force: Skip hash comparison and always re-index.
      """
      from memorymesh.server.auth_guard import check_access
  
      if (err := check_access(ctx, "index")) is not None:
          return err
  
      t0 = time.perf_counter()
      errors: list[str] = []
      n_files = 0
      n_chunks = 0
  
      if path is not None:
          target = Path(path)
          if not target.exists():
              return {"error": f"Path does not exist: {path}"}
  
          if target.is_file():
              result = ctx.indexer.index_file(target, force=force)
              n_files = 1
              n_chunks = result.n_chunks
              if result.error:
                  errors.append(f"{path}: {result.error}")
          else:
              results = ctx.indexer.index_directory(target, force=force)
              n_files = len(results)
              n_chunks = sum(r.n_chunks for r in results)
              errors = [f"{r.path}: {r.error}" for r in results if r.error]
      else:
          for src in ctx.config.sources:
              src_name = src.name or str(src.path)
              if not src.path.exists():
                  errors.append(f"Source path not found: {src.path}")
                  continue
              results = ctx.indexer.index_directory(
                  src.path,
                  source_name=src_name,
                  recursive=src.recursive,
                  extensions=src.extensions or None,
                  ignore_patterns=src.ignore or None,
                  force=force,
              )
              n_files += len(results)
              n_chunks += sum(r.n_chunks for r in results)
              errors += [f"{r.path}: {r.error}" for r in results if r.error]
  
      latency_ms = (time.perf_counter() - t0) * 1000
      logger.info(
          f"index_now: files={n_files} chunks={n_chunks} "
          f"errors={len(errors)} duration={latency_ms:.0f}ms"
      )
      ctx.audit_logger.log_query(
          tool="index_now",
          query=path or "<all sources>",
          n_results=n_files,
          latency_ms=latency_ms,
      )
  
      return {
          "n_files_processed": n_files,
          "n_chunks": n_chunks,
          "duration_ms": round(latency_ms, 2),
          "errors": errors,
      }

• src/memorymesh/server/app.py:115-129 (registration)

  Import of the index_now module and registration call index_now.register(mcp, ctx) which wires the tool onto the FastMCP server.

      index_now,
      list_sources,
      pin_memory,
      query_timeline,
      related_documents,
      search_by_date,
      search_memory,
      summarize_source,
      sync_source,
  )
  
  search_memory.register(mcp, ctx)
  list_sources.register(mcp, ctx)
  get_document.register(mcp, ctx)
  index_now.register(mcp, ctx)

• src/memorymesh/core/models/search.py:96-102 (schema)

  The IndexResponse model defining the schema of the tool's return value (n_files_processed, n_chunks, duration_ms, errors).

  class IndexResponse(BaseModel):
      """Full response from the :func:`index_now` MCP tool."""
  
      n_files_processed: int
      n_chunks: int
      duration_ms: float
      errors: list[str] = Field(default_factory=list)

• src/memorymesh/server/tools/index_now.py:20-28 (helper)

  The register() function that takes (mcp, ctx) and registers index_now as a FastMCP tool via the @mcp.tool() decorator (closure injection pattern).

  def register(mcp: FastMCP, ctx: AppContext) -> None:
      """Register the ``index_now`` tool on *mcp* with *ctx* injected.
  
      Args:
          mcp: The FastMCP instance to register onto.
          ctx: Shared application context (injected via closure).
      """
  
      @mcp.tool()