Roam Research MCP Server

Instructions

Implementation Reference

• main.py:389-417 (handler)

  The primary MCP tool handler for 'get_page_content'. Registered via @mcp.tool() decorator. Fetches page content using the Roam client and serializes to JSON string.
  @mcp.tool() async def get_page_content(page_name: str) -> str: """Get the complete content of a specific page in Roam Research with all nested child blocks. Retrieves all blocks on the specified page with their hierarchical structure, including nested children up to 5 levels deep. Returns content in markdown format with proper indentation to reflect the block hierarchy. Args: page_name: Exact name of the page to retrieve (case-sensitive) Returns: JSON string containing: - result: Array of blocks with content and timestamps - Each block includes: content (markdown), timestamp (edit time) Examples: get_page_content("Daily Notes") get_page_content("Project Planning") get_page_content("信用卡") """ try: client = get_roam_client() result = client.get_page_content(page_name) return json.dumps(result, indent=2) except Exception as e: print(f"Error getting page content: {e}", file=sys.stderr) return f"Error: {str(e)}"
• main.py:85-129 (helper)

  Core helper method in RoamResearchMCPServer class that implements the Datalog query to fetch page blocks with nested children, sorts by edit time, converts to markdown hierarchy, and returns structured result.
  def get_page_content(self, page_name: str) -> Dict[str, Any]: """Get content of a specific page with child blocks""" # Query to get all blocks on the page with nested children query = """[:find (pull ?block [:block/string :block/uid :edit/time :block/order {:block/children [:block/string :block/uid :edit/time {:block/children [:block/string :block/uid :edit/time {:block/children [:block/string :block/uid :edit/time {:block/children [:block/string :block/uid :edit/time]}]}]}]}]) ?time :in $ ?PAGE :where [?page :node/title ?PAGE] [?block :block/page ?page] [?block :edit/time ?time] ]""" data = {"query": query, "args": [page_name]} endpoint = f"/api/graph/{self.graph_name}/q" raw_result = self._make_request("POST", endpoint, data) # Sort by time (descending) results = raw_result.get("result", []) sorted_results = sorted(results, key=lambda x: x[1], reverse=True) # Transform the result to only include markdown content with children simplified_result = [] for item in sorted_results: if item and len(item) > 0: block = item[0] timestamp = item[1] content = self._build_block_with_children(block) simplified_result.append({"content": content, "timestamp": timestamp}) return {"result": simplified_result}
• main.py:68-84 (helper)

  Recursive helper function to construct hierarchical markdown from Roam blocks, including content conversion and child indentation.
  def _build_block_with_children(self, block: Dict[str, Any]) -> str: """Build a markdown string with block content and all its children""" content = self._convert_block_to_markdown(block) # Get children from the nested data structure children = block.get(':block/children', []) if children: content += "\n" for child in children: child_content = self._build_block_with_children(child) # Indent child content child_lines = child_content.split('\n') indented_lines = [' ' + line for line in child_lines if line.strip()] content += '\n'.join(indented_lines) + '\n' return content.strip()
• main.py:58-66 (helper)

  Utility helper to convert Roam block string content to markdown format, specifically handling [[page]] link syntax.
  def _convert_block_to_markdown(self, block: Dict[str, Any]) -> str: """Convert a Roam block to markdown format""" content = block.get(':block/string', '') # Convert Roam-style links [[page]] to markdown links content = re.sub(r'\[\[([^\]]+)\]\]', r'[\1](\1)', content) return content
• main.py:369-386 (helper)

  Singleton helper to lazily initialize and retrieve the RoamResearchMCPServer client instance using environment variables.
  def get_roam_client(): """Get or initialize Roam Research client""" global roam_client if roam_client is None: token = os.getenv("ROAM_TOKEN") graph_name = os.getenv("ROAM_GRAPH_NAME") print(f"DEBUG: ROAM_TOKEN present: {bool(token)}", file=sys.stderr) print(f"DEBUG: ROAM_GRAPH_NAME present: {bool(graph_name)}", file=sys.stderr) if not token or not graph_name: raise Exception("ROAM_TOKEN and ROAM_GRAPH_NAME environment variables are required") roam_client = RoamResearchMCPServer(token, graph_name) print("DEBUG: Roam client initialized successfully", file=sys.stderr) return roam_client