Mnemosyne MCP

Instructions

Implementation Reference

• src/neem/mcp/tools/hocuspocus.py:909-976 (handler)

  MCP tool handler for 'query_blocks'. Authenticates, connects to Hocuspocus document channel, invokes DocumentReader.query_blocks with filters, and returns matching blocks.
  name="query_blocks", title="Query Blocks", description=( "Search for blocks matching specific criteria. Filter by block type, indent level, " "list type, checked state, or text content. Returns a list of matching block summaries. " "Use this to find blocks without reading the entire document." ), ) async def query_blocks_tool( graph_id: str, document_id: str, block_type: Optional[str] = None, indent: Optional[int] = None, indent_gte: Optional[int] = None, indent_lte: Optional[int] = None, list_type: Optional[str] = None, checked: Optional[bool] = None, text_contains: Optional[str] = None, limit: int = 50, context: Context | None = None, ) -> dict: """Query blocks matching specific criteria.""" auth = MCPAuthContext.from_context(context) auth.require_auth() if not graph_id or not graph_id.strip(): raise ValueError("graph_id is required") if not document_id or not document_id.strip(): raise ValueError("document_id is required") try: await hp_client.connect_document(graph_id.strip(), document_id.strip()) channel = hp_client.get_document_channel(graph_id.strip(), document_id.strip()) if channel is None: raise RuntimeError(f"Document channel not found: {graph_id}/{document_id}") reader = DocumentReader(channel.doc) matches = reader.query_blocks( block_type=block_type, indent=indent, indent_gte=indent_gte, indent_lte=indent_lte, list_type=list_type, checked=checked, text_contains=text_contains, limit=limit, ) result = { "graph_id": graph_id.strip(), "document_id": document_id.strip(), "count": len(matches), "blocks": matches, } return result except Exception as e: logger.error( "Failed to query blocks", extra_context={ "graph_id": graph_id, "document_id": document_id, "error": str(e), }, ) raise RuntimeError(f"Failed to query blocks: {e}")
• src/neem/hocuspocus/document.py:416-506 (helper)

  DocumentReader.query_blocks: Iterates over Y.XmlFragment children (blocks), applies filters on type, indent levels, list_type, checked status, and text content using regex, collects summaries up to limit.
  def query_blocks( self, block_type: str | None = None, indent: int | None = None, indent_gte: int | None = None, indent_lte: int | None = None, list_type: str | None = None, checked: bool | None = None, text_contains: str | None = None, limit: int = 50, ) -> list[dict[str, Any]]: """Query blocks matching specific criteria. Args: block_type: Filter by block type (paragraph, heading, listItem, etc.) indent: Filter by exact indent level indent_gte: Filter by indent >= value indent_lte: Filter by indent <= value list_type: For listItems, filter by listType (bullet, ordered, task) checked: For task items, filter by checked state text_contains: Filter by text content containing this string limit: Maximum number of results to return Returns: List of matching block summaries. """ fragment = self.get_content_fragment() matches = [] import re for i, child in enumerate(fragment.children): if len(matches) >= limit: break if not hasattr(child, "attributes"): continue attrs = dict(child.attributes) tag = child.tag if hasattr(child, "tag") else "unknown" # Filter by type if block_type and tag != block_type: continue # Filter by indent elem_indent = attrs.get("indent", 0) if isinstance(elem_indent, str): try: elem_indent = int(elem_indent) except ValueError: elem_indent = 0 if indent is not None and elem_indent != indent: continue if indent_gte is not None and elem_indent < indent_gte: continue if indent_lte is not None and elem_indent > indent_lte: continue # Filter by listType if list_type and attrs.get("listType") != list_type: continue # Filter by checked if checked is not None: elem_checked = attrs.get("checked", False) if isinstance(elem_checked, str): elem_checked = elem_checked.lower() == "true" if elem_checked != checked: continue # Filter by text content text = str(child) plain_text = re.sub(r"<[^>]+>", "", text).strip() if text_contains and text_contains.lower() not in plain_text.lower(): continue # Build match summary matches.append({ "block_id": attrs.get("data-block-id"), "index": i, "type": tag, "text_preview": plain_text[:100] + ("..." if len(plain_text) > 100 else ""), "attributes": { k: v for k, v in attrs.items() if k not in ("data-block-id",) # Exclude redundant fields }, }) return matches
• src/neem/mcp/server/standalone_server.py:317-317 (registration)

  Registration call for hocuspocus tools including query_blocks during standalone MCP server setup.
  register_hocuspocus_tools(mcp_server)