brain-mcp

Instructions

    Search markdown corpus and IP documents with various filters.

    Args:
        query: Search query (keyword for markdown, semantic for IP docs)
        filter: What to search/filter:
            - None (default): Keyword search on markdown corpus
            - "ip": Vector search on curated IP documents
            - "breakthrough": Documents with BREAKTHROUGH energy
            - "deep": High depth-score documents
            - "project": Documents for a specific project
            - "todos": Documents with open TODOs
        project: Project name (used with filter="project" or filter="todos")
        limit: Max results (default 15)
        min_depth: Minimum depth score (used with filter="deep", default 70)
    

Implementation Reference

• brain_mcp/server/tools_search.py:450-605 (handler)

  The `search_docs` function implements the tool, providing search functionality across markdown documents and IP documents with various filter options.

  def search_docs(query: str = "", filter: str = None, project: str = None,
                  limit: int = 15, min_depth: int = 70) -> str:
      """
      Search markdown corpus and IP documents with various filters.
  
      Args:
          query: Search query (keyword for markdown, semantic for IP docs)
          filter: What to search/filter:
              - None (default): Keyword search on markdown corpus
              - "ip": Vector search on curated IP documents
              - "breakthrough": Documents with BREAKTHROUGH energy
              - "deep": High depth-score documents
              - "project": Documents for a specific project
              - "todos": Documents with open TODOs
          project: Project name (used with filter="project" or filter="todos")
          limit: Max results (default 15)
          min_depth: Minimum depth score (used with filter="deep", default 70)
      """
      cfg = get_config()
  
      if filter == "ip":
          return _search_ip_docs(query, limit)
      elif filter == "breakthrough":
          return _get_breakthrough_docs(limit)
      elif filter == "deep":
          return _get_deep_docs(min_depth, limit)
      elif filter == "project":
          return _get_project_docs(project or query, limit)
      elif filter == "todos":
          return _get_open_todos(project, limit)
  
      # Default: keyword search
      db = get_markdown_db()
      if not db:
          return "Markdown corpus not found. Ingest markdown docs first."
  
      pattern = f"%{query}%"
      results = db.execute("""
          SELECT filename, project, voice, energy, depth_score, harvest_score,
                 decision_count, word_count, first_line
          FROM markdown_docs
          WHERE content ILIKE ? OR title ILIKE ? OR filename ILIKE ?
          ORDER BY depth_score DESC, harvest_score DESC
          LIMIT ?
      """, [pattern, pattern, pattern, limit]).fetchall()
  
      if not results:
          return f"No markdown documents found for '{query}'"
  
      output = [f"## Markdown Search: '{query}'\n", f"_Found {len(results)} documents_\n"]
      for r in results:
          fname, proj, voice, energy, depth, harvest, decisions, words, preview = r
          output.append(f"**{fname}**")
          output.append(f"  Project: {proj or 'unassigned'} | Voice: {voice} | Energy: {energy}")
          output.append(f"  Depth: {depth} | Harvest: {harvest} | Decisions: {decisions} | Words: {words:,}")
          output.append(f"  > {preview[:100]}...\n")
  
      return "\n".join(output)
  
  def _search_ip_docs(query: str, limit: int = 10) -> str:
      """Vector search on curated IP documents."""
      cfg = get_config()
      if not cfg.lance_path.exists():
          return "Vector database not found."
  
      embedding = get_embedding(query)
      if not embedding:
          return "Could not generate embedding."
  
      db = get_lance_db()
      if not db:
          return "Could not connect to vector database."
  
      try:
          tbl = db.open_table("markdown")
          results = tbl.search(embedding).limit(limit).to_pandas()
      except Exception as e:
          return f"Search error: {e}"
  
      if results.empty:
          return "No IP documents found."
  
      output = [f"## IP Document Search: '{query}'\n"]
      for i, row in results.iterrows():
          filename = row.get("filename", "Unknown")
          ip_type = row.get("ip_type", "unknown")
          depth = row.get("depth_score", 0)
          energy = row.get("energy", "unknown")
          words = row.get("word_count", 0)
          preview = row.get("content_preview", "")[:300]
          distance = row.get("_distance", 0)
  
          output.append(f"### {i+1}. {filename}")
          output.append(f"**Type**: {ip_type} | **Depth**: {depth} | **Energy**: {energy} | **Words**: {words:,}")
          output.append(f"**Similarity**: {distance:.4f}")
          output.append(f"> {preview}...\n")
  
      return "\n".join(output)
  
  def _get_breakthrough_docs(limit: int = 20) -> str:
      db = get_markdown_db()
      if not db:
          return "Markdown corpus not found."
      results = db.execute("""
          SELECT filename, project, depth_score, harvest_score, decision_count,
                 word_count, first_line
          FROM markdown_docs WHERE energy = 'BREAKTHROUGH'
          ORDER BY depth_score DESC, harvest_score DESC LIMIT ?
      """, [limit]).fetchall()
      if not results:
          return "No breakthrough documents found."
      output = [f"## Breakthrough Documents\n"]
      for fname, proj, depth, harvest, decisions, words, preview in results:
          output.append(f"**{fname}** (depth: {depth}, harvest: {harvest})")
          output.append(f"  Project: {proj or 'unassigned'} | Decisions: {decisions} | Words: {words:,}")
          output.append(f"  > {preview[:120]}...\n")
      return "\n".join(output)
  
  def _get_deep_docs(min_depth: int = 70, limit: int = 20) -> str:
      db = get_markdown_db()
      if not db:
          return "Markdown corpus not found."
      results = db.execute("""
          SELECT filename, project, voice, energy, depth_score, harvest_score,
                 decision_count, seed_concepts, word_count
          FROM markdown_docs WHERE depth_score >= ?
          ORDER BY depth_score DESC LIMIT ?
      """, [min_depth, limit]).fetchall()
      if not results:
          return f"No documents with depth >= {min_depth}"
      output = [f"## Deep Documents (depth >= {min_depth})\n"]
      for fname, proj, voice, energy, depth, _, decisions, seeds, words in results:
          output.append(f"**{fname}** (depth: {depth})")
          output.append(f"  Project: {proj or 'unassigned'} | Voice: {voice} | Energy: {energy}")
          output.append(f"  Decisions: {decisions} | Words: {words:,}\n")
      return "\n".join(output)
  
  def _get_project_docs(project: str, limit: int = 20) -> str:
      db = get_markdown_db()
      if not db:
          return "Markdown corpus not found."
      results = db.execute("""
          SELECT filename, voice, energy, depth_score, harvest_score,
                 decision_count, todos_open, word_count, first_line
          FROM markdown_docs WHERE project = ?
          ORDER BY depth_score DESC LIMIT ?
      """, [project.lower(), limit]).fetchall()
      if not results:
          return f"No documents found for project '{project}'"
      output = [f"## Project: {project}\n"]
      for fname, voice, energy, depth, harvest, decisions, todos, _, preview in results:
          output.append(f"**{fname}** (depth: {depth}, harvest: {harvest})")
          output.append(f"  Voice: {voice} | Energy: {energy} | Decisions: {decisions} | TODOs: {todos}")
          output.append(f"  > {preview[:100]}...\n")
      return "\n".join(output)