list_sources

Instructions

    Returns a summary for each source: file counts by status, total chunk
    count, and aggregate disk size.
    

Implementation Reference

• src/memorymesh/server/tools/list_sources.py:26-76 (handler)

  The MCP tool handler for list_sources. Iterates over configured sources, queries the metadata store for file records, computes indexing statistics (indexed/pending/errored counts, chunk count, disk size, last scan), logs the query, and returns the aggregated results.

  @mcp.tool()
  def list_sources() -> dict:
      """List all configured source directories with indexing statistics.
  
      Returns a summary for each source: file counts by status, total chunk
      count, and aggregate disk size.
      """
      from memorymesh.server.auth_guard import check_access
  
      if (err := check_access(ctx, "read")) is not None:
          return err
  
      t0 = time.perf_counter()
  
      sources_out = []
      for src_cfg in ctx.config.sources:
          name = src_cfg.name or str(src_cfg.path)
  
          all_records = ctx.metadata_store.list_files()
          matching = [r for r in all_records if r.source_name == name]
  
          n_indexed = sum(1 for r in matching if r.status == "indexed")
          n_pending = sum(1 for r in matching if r.status == "pending_reindex")
          n_errored = sum(1 for r in matching if r.status == "parse_error")
          total_chunks = sum(r.n_chunks for r in matching if r.status == "indexed")
          disk_bytes = sum(r.size_bytes for r in matching)
          last_scan = max((r.indexed_at for r in matching), default=None)
  
          sources_out.append(
              {
                  "name": name,
                  "path": str(src_cfg.path),
                  "recursive": src_cfg.recursive,
                  "n_files_indexed": n_indexed,
                  "n_files_pending": n_pending,
                  "n_files_errored": n_errored,
                  "total_chunks": total_chunks,
                  "disk_size_bytes": disk_bytes,
                  "last_scan_at": last_scan,
              }
          )
  
      latency_ms = (time.perf_counter() - t0) * 1000
      ctx.audit_logger.log_query(
          tool="list_sources",
          query="list_sources",
          n_results=len(sources_out),
          latency_ms=latency_ms,
      )
  
      return {"sources": sources_out, "duration_ms": round(latency_ms, 2)}

• src/memorymesh/server/app.py:126-127 (registration)

  Registration of the list_sources tool on the FastMCP instance via its register function.

  search_memory.register(mcp, ctx)
  list_sources.register(mcp, ctx)

• src/memorymesh/storage/file_repository.py:231-234 (helper)

  Database-backed helper method in FileRepository that lists all source rows from the 'sources' table.

  def list_sources(self) -> list[dict[str, object]]:
      """Return all source rows as plain dicts."""
      rows = self._conn().execute("SELECT * FROM sources").fetchall()
      return [dict(r) for r in rows]

• src/memorymesh/storage/metadata_store.py:455-457 (helper)

  Delegation method in MetadataStore that calls through to FileRepository.list_sources().

  def list_sources(self) -> list[dict[str, object]]:
      """Return all source rows as plain dicts."""
      return self._files.list_sources()