Roam Research MCP Server

Instructions

Implementation Reference

• main.py:453-483 (handler)

  MCP tool handler: @mcp.tool()-decorated async function that executes the write_to_page tool by calling the Roam client method and handles errors.
  @mcp.tool() async def write_to_page(page_name: str, content: str) -> str: """Write hierarchical markdown content to a specific page in Roam Research. Creates a new block structure on the specified page with automatic indentation detection. Supports nested bullet points and maintains proper parent-child relationships between blocks. Content is appended to the end of the page. Args: page_name: Exact name of the target page (case-sensitive, must exist) content: Hierarchical markdown content using '- ' prefix and indentation Format: "- Main topic\n - Subtopic\n - Details" Supports any indentation level with automatic detection Returns: JSON string containing: - result: "success" if completed - blocks_created: Total number of blocks created (including children) - details: Array of individual block creation results Examples: write_to_page("Project Notes", "- New milestone\n - Task 1\n - Task 2") write_to_page("信用卡", "- [[銀行/國泰]]\n - 現金回饋 2%") """ try: client = get_roam_client() result = client.write_to_page(page_name, content) return f"Successfully wrote to page '{page_name}': {json.dumps(result, indent=2)}" except Exception as e: print(f"Error writing to page: {e}", file=sys.stderr) return f"Error: {str(e)}"
• main.py:211-239 (helper)

  Core implementation in RoamResearchMCPServer class: retrieves page UID, parses markdown content into blocks, creates hierarchical structure via API.
  def write_to_page(self, page_name: str, content: str) -> Dict[str, Any]: """Write hierarchical content to a specific page""" # First, get the page UID page_query = f"""[:find ?uid :in $ ?PAGE :where [?e :node/title ?PAGE] [?e :block/uid ?uid] ]""" query_data = {"query": page_query, "args": [page_name]} endpoint = f"/api/graph/{self.graph_name}/q" page_result = self._make_request("POST", endpoint, query_data) # Get page UID if not page_result.get("result") or not page_result["result"]: raise ValueError(f"Page '{page_name}' not found") page_uid = page_result["result"][0][0] # Parse markdown content into hierarchical blocks blocks = self._parse_markdown_to_blocks(content) # Create the hierarchical structure results = self._create_block_hierarchy(page_uid, blocks) return {"result": "success", "blocks_created": len(results), "details": results}
• main.py:240-301 (helper)

  Helper function that parses hierarchical markdown content into a nested block structure with UIDs.
  def _parse_markdown_to_blocks(self, content: str) -> list: """Parse markdown content into hierarchical block structure using dynamic indentation detection""" lines = [line.rstrip() for line in content.split('\n') if line.strip()] # Build indentation level mapping indent_map = {} # {actual_indent: level} blocks = [] stack = [] # Stack to track parent blocks at different levels for i, line in enumerate(lines): actual_indent = len(line) - len(line.lstrip()) # Determine the level for this indentation if actual_indent not in indent_map: if actual_indent == 0: indent_map[actual_indent] = 0 else: # Find the closest parent indentation level parent_indents = [k for k in indent_map.keys() if k < actual_indent] if parent_indents: parent_indent = max(parent_indents) indent_map[actual_indent] = indent_map[parent_indent] + 1 else: indent_map[actual_indent] = 1 level = indent_map[actual_indent] # Extract content (remove leading "- " if present) text = line.strip() if text.startswith('- '): text = text[2:] # Create block structure block = { "string": text, "uid": f"{datetime.now().strftime('%m-%d-%Y')}-{datetime.now().strftime('%H%M%S')}-{i}", "children": [] } # Adjust stack to current level while len(stack) > level: stack.pop() if level == 0: # Top-level block blocks.append(block) stack = [block] else: # Child block - add to the parent at the appropriate level if stack and len(stack) >= level: parent = stack[level - 1] parent["children"].append(block) # Extend stack to current level while len(stack) <= level: stack.append(block) stack[level] = block else: # Fallback: treat as top-level if stack is insufficient blocks.append(block) stack = [block] return blocks
• main.py:303-327 (helper)

  Recursive helper that creates blocks and their children via Roam API write endpoint.
  def _create_block_hierarchy(self, parent_uid: str, blocks: list) -> list: """Recursively create blocks with their children""" results = [] for block in blocks: # Create the main block block_data = { "action": "create-block", "location": {"parent-uid": parent_uid, "order": "last"}, "block": { "string": block["string"], "uid": block["uid"], }, } write_endpoint = f"/api/graph/{self.graph_name}/write" result = self._make_request("POST", write_endpoint, block_data) results.append(result) # Create children if they exist if block["children"]: child_results = self._create_block_hierarchy(block["uid"], block["children"]) results.extend(child_results) return results
• main.py:453-454 (registration)

  MCP tool registration decorator for write_to_page.
  @mcp.tool() async def write_to_page(page_name: str, content: str) -> str: