insert_block
Insert a new content block before or after an existing block in Mnemosyne knowledge graphs. Specify position and XML content to add structured data.
Instructions
Insert a new block relative to an existing block. Use position='after' or 'before' to specify where to insert. Returns the new block's generated ID. For appending to the end, use append_to_document instead.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| graph_id | Yes | ||
| document_id | Yes | ||
| reference_block_id | Yes | ||
| xml_content | Yes | ||
| position | No | after |
Implementation Reference
- src/neem/mcp/tools/hocuspocus.py:1061-1069 (registration)Registers the MCP tool named 'insert_block' with input schema defined by function parameters (graph_id, document_id, reference_block_id, xml_content, position) and description.@server.tool( name="insert_block", title="Insert Block", description=( "Insert a new block relative to an existing block. Use position='after' or 'before' " "to specify where to insert. Returns the new block's generated ID. " "For appending to the end, use append_to_document instead." ), )
- Handler function for 'insert_block' tool. Authenticates, connects to document via HocuspocusClient, performs transaction using DocumentWriter.insert_block_after_id or before_id based on position, reads back new block info, returns success with new_block_id.async def insert_block_tool( graph_id: str, document_id: str, reference_block_id: str, xml_content: str, position: str = "after", context: Context | None = None, ) -> dict: """Insert a new block before or after a reference block. Args: graph_id: The graph containing the document document_id: The document to insert into reference_block_id: The block to insert relative to xml_content: TipTap XML for the new block position: 'after' or 'before' the reference block """ 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") if not reference_block_id or not reference_block_id.strip(): raise ValueError("reference_block_id is required") if not xml_content or not xml_content.strip(): raise ValueError("xml_content is required") if position not in ("after", "before"): raise ValueError("position must be 'after' or 'before'") try: await hp_client.connect_document(graph_id.strip(), document_id.strip()) new_block_id: str = "" def perform_insert(doc: Any) -> None: nonlocal new_block_id writer = DocumentWriter(doc) if position == "after": new_block_id = writer.insert_block_after_id( reference_block_id.strip(), xml_content.strip() ) else: new_block_id = writer.insert_block_before_id( reference_block_id.strip(), xml_content.strip() ) await hp_client.transact_document( graph_id.strip(), document_id.strip(), perform_insert, ) # Read back the new block 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) block_info = reader.get_block_info(new_block_id) if new_block_id else None result = { "success": True, "graph_id": graph_id.strip(), "document_id": document_id.strip(), "new_block_id": new_block_id, "block": block_info, } return result except Exception as e: logger.error( "Failed to insert block", extra_context={ "graph_id": graph_id, "document_id": document_id, "reference_block_id": reference_block_id, "error": str(e), }, ) raise RuntimeError(f"Failed to insert block: {e}")
- Core helper method in DocumentWriter that finds the reference block by ID, parses input XML, generates new block ID if needed, processes XML to Y.js elements (handling lists etc.), inserts after the reference index in transaction.def insert_block_after_id(self, after_block_id: str, xml_str: str) -> str: """Insert a new block after the specified block. Args: after_block_id: The block ID to insert after xml_str: TipTap XML for the new block Returns: The new block's generated ID. Raises: ValueError: If reference block not found. """ logger.info( "insert_block_after_id: starting", extra_context={ "after_block_id": after_block_id, "xml_str": xml_str[:200], }, ) result = self.find_block_by_id(after_block_id) if result is None: logger.error( "insert_block_after_id: reference block not found", extra_context={"after_block_id": after_block_id}, ) raise ValueError(f"Block not found: {after_block_id}") index, _ = result fragment = self.get_content_fragment() block_count_before = len(list(fragment.children)) elem = ET.fromstring(xml_str) # Pre-generate block ID if not already set new_block_id = elem.get("data-block-id") if not new_block_id: new_block_id = _generate_block_id() elem.set("data-block-id", new_block_id) logger.info( "insert_block_after_id: inserting at position", extra_context={ "insert_after_index": index, "new_block_id": new_block_id, "elem_tag": elem.tag, }, ) with self._doc.transaction(): blocks = self._process_element(elem) logger.info( "insert_block_after_id: processed into blocks", extra_context={ "num_blocks": len(blocks), "source_tag": elem.tag, }, ) for i, block in enumerate(blocks): fragment.children.insert(index + 1 + i, block) self._apply_pending_formats() block_count_after = len(list(fragment.children)) logger.info( "insert_block_after_id: completed", extra_context={ "new_block_id": new_block_id, "block_count_before": block_count_before, "block_count_after": block_count_after, "content_after": str(fragment)[:500], }, ) return new_block_id
- Core helper method in DocumentWriter similar to insert_block_after_id but inserts before the reference block index.def insert_block_before_id(self, before_block_id: str, xml_str: str) -> str: """Insert a new block before the specified block. Args: before_block_id: The block ID to insert before xml_str: TipTap XML for the new block Returns: The new block's generated ID. Raises: ValueError: If reference block not found. """ result = self.find_block_by_id(before_block_id) if result is None: raise ValueError(f"Block not found: {before_block_id}") index, _ = result fragment = self.get_content_fragment() elem = ET.fromstring(xml_str) # Pre-generate block ID if not already set new_block_id = elem.get("data-block-id") if not new_block_id: new_block_id = _generate_block_id() elem.set("data-block-id", new_block_id) with self._doc.transaction(): blocks = self._process_element(elem) for i, block in enumerate(blocks): fragment.children.insert(index + i, block) self._apply_pending_formats() return new_block_id