zettelkasten-mcp

Instructions

        Args:
            limit: Maximum number of results to return (default: 10)
        

Implementation Reference

• src/zettelkasten_mcp/server/mcp_server.py:606-639 (handler)

  The MCP tool handler function 'zk_find_central_notes' that retrieves central notes from the search service and formats the output as a ranked list with previews.

  def zk_find_central_notes(limit: int = 10) -> str:
      """Identify the most connected notes that serve as knowledge hubs.
  
      Notes are ranked by their total number of connections (incoming + outgoing links),
      determining their centrality in the knowledge network. These central notes often
      represent key concepts or structure notes.
  
      Args:
          limit: Maximum number of central notes to return (default: 10)
      """
      try:
          # Get central notes
          central_notes = self.search_service.find_central_notes(limit)
          if not central_notes:
              return "No notes found with connections."
  
          # Format results
          output = "Central notes in the Zettelkasten (most connected):\n\n"
          for i, (note, connection_count) in enumerate(central_notes, 1):
              output += f"{i}. {note.title} (ID: {note.id})\n"
              output += f"   Connections: {connection_count}\n"
              if note.tags:
                  output += (
                      f"   Tags: {', '.join(tag.name for tag in note.tags)}\n"
                  )
              # Add a snippet of content (first 100 chars)
              content_preview = note.content[:100].replace("\n", " ")
              if len(note.content) > 100:
                  content_preview += "..."
              output += f"   Preview: {content_preview}\n\n"
          return output
      except Exception as e:
          return self.format_error_response(e)

• src/zettelkasten_mcp/server/mcp_server.py:597-605 (registration)

  The @mcp.tool decorator that registers the 'zk_find_central_notes' tool with its metadata and annotations.

  @self.mcp.tool(
      name="zk_find_central_notes",
      description="Identify the most connected notes that serve as knowledge hubs.",
      annotations={
          "readOnlyHint": True,
          "destructiveHint": False,
          "idempotentHint": True,
      },
  )

• src/zettelkasten_mcp/services/search_service.py:159-202 (helper)

  The SearchService.find_central_notes method containing the core database query logic to compute note centrality based on link counts and return the top connected notes.

  def find_central_notes(self, limit: int = 10) -> List[Tuple[Note, int]]:
      """Find notes with the most connections (incoming + outgoing links)."""
      note_connections = []
      # Direct database query to count connections for all notes at once
      with self.zettel_service.repository.session_factory() as session:
          # Use a CTE for better readability and performance
          query = text("""
          WITH outgoing AS (
              SELECT source_id as note_id, COUNT(*) as outgoing_count 
              FROM links 
              GROUP BY source_id
          ),
          incoming AS (
              SELECT target_id as note_id, COUNT(*) as incoming_count 
              FROM links 
              GROUP BY target_id
          )
          SELECT n.id,
              COALESCE(o.outgoing_count, 0) as outgoing,
              COALESCE(i.incoming_count, 0) as incoming,
              (COALESCE(o.outgoing_count, 0) + COALESCE(i.incoming_count, 0)) as total
          FROM notes n
          LEFT JOIN outgoing o ON n.id = o.note_id
          LEFT JOIN incoming i ON n.id = i.note_id
          WHERE (COALESCE(o.outgoing_count, 0) + COALESCE(i.incoming_count, 0)) > 0
          ORDER BY total DESC
          LIMIT :limit
          """)
          
          results = session.execute(query, {"limit": limit}).all()
          
          # Process results
          for note_id, outgoing_count, incoming_count, total_connections in results:
              total_connections = outgoing_count + incoming_count
              if total_connections > 0:  # Only include notes with connections
                  note = self.zettel_service.get_note(note_id)
                  if note:
                      note_connections.append((note, total_connections))
      
      # Sort by total connections (descending)
      note_connections.sort(key=lambda x: x[1], reverse=True)
      
      # Return top N notes
      return note_connections[:limit]