knowledge-rag

Instructions

Add a new document to the knowledge base from raw content.

Saves the content to the documents directory and indexes it immediately.

Args:
    content: Full text content of the document
    filepath: Relative path within documents dir (e.g., "security/new-technique.md")
    category: Document category (security, ctf, logscale, development, general)

Returns:
    JSON string with indexing results

Implementation Reference

• mcp_server/server.py:1422-1448 (handler)

  MCP tool registration and wrapper for `add_document` which calls the orchestrator.

  @mcp.tool()
  def add_document(content: str, filepath: str, category: str = "general") -> str:
      """
      Add a new document to the knowledge base from raw content.
  
      Saves the content to the documents directory and indexes it immediately.
  
      Args:
          content: Full text content of the document
          filepath: Relative path within documents dir (e.g., "security/new-technique.md")
          category: Document category (security, ctf, logscale, development, general)
  
      Returns:
          JSON string with indexing results
      """
      if not content or not content.strip():
          return json.dumps({"status": "error", "message": "Content cannot be empty"})
      if not filepath or not filepath.strip():
          return json.dumps({"status": "error", "message": "Filepath cannot be empty"})
  
      orchestrator = get_orchestrator()
      result = orchestrator.add_document_from_content(content.strip(), filepath.strip(), category)
  
      if "error" in result:
          return json.dumps({"status": "error", "message": result["error"]})
  
      return json.dumps({"status": "success", **result}, indent=2)

• mcp_server/server.py:937-980 (handler)

  Core logic for adding a document from content, saving it to disk, and updating the index.

  def add_document_from_content(self, content: str, filepath: str, category: str) -> Dict[str, Any]:
      """Add a new document from raw content string. Saves to disk and indexes."""
      full_path = config.documents_dir / filepath
      full_path.parent.mkdir(parents=True, exist_ok=True)
      full_path.write_text(content, encoding="utf-8")
  
      doc = self.parser.parse_file(full_path)
      if not doc:
          return {"error": "Failed to parse document content"}
  
      doc.category = category
      for chunk in doc.chunks:
          chunk.metadata["category"] = category
  
      chunks_added, dedup_skipped = self._index_document(doc)
  
      try:
          file_stat = full_path.stat()
          file_mtime = datetime.fromtimestamp(file_stat.st_mtime).isoformat()
          file_size = file_stat.st_size
      except OSError:
          file_mtime = datetime.now().isoformat()
          file_size = 0
  
      self._indexed_docs[doc.id] = {
          "source": str(full_path),
          "category": category,
          "format": doc.format,
          "chunks": chunks_added,
          "keywords": doc.keywords,
          "indexed_at": datetime.now().isoformat(),
          "file_mtime": file_mtime,
          "file_size": file_size,
      }
      self._save_metadata()
      self.query_cache.invalidate()
      self.bm25_index.build_index()
  
      return {
          "chunks_added": chunks_added,
          "dedup_skipped": dedup_skipped,
          "category": category,
          "filepath": str(full_path),
      }