zk_find_central_notes
Identify key notes in your Zettelkasten system by ranking them based on total incoming and outgoing links, highlighting central nodes in the knowledge network.
Instructions
Find notes with the most connections (incoming + outgoing links). Notes are ranked by their total number of connections, determining their centrality in the knowledge network. Due to database constraints, only one link of each type is counted between any pair of notes.
Args:
limit: Maximum number of results to return (default: 10)
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| limit | No |
Implementation Reference
- 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, }, )
- 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]