Roam Research MCP Server

Instructions

Implementation Reference

• main.py:419-450 (handler)

  MCP tool handler for 'get_page_references'. Decorated with @mcp.tool(), includes schema in docstring and type hints. Calls internal client method and serializes to JSON.
  @mcp.tool() async def get_page_references(page_name: str, limit: int = 10, cursor: Optional[int] = None) -> str: """Get all blocks that reference a specific page in Roam Research with pagination support. Finds all blocks across your Roam database that contain links to the specified page. Results are sorted by most recent edit time and include the full hierarchical context of each referencing block. Args: page_name: Exact name of the page to find references for (case-sensitive) limit: Maximum number of references to return per request (default: 10) cursor: Timestamp cursor for pagination - use next_cursor from previous response to get additional results Returns: JSON string containing: - result: Array of referencing blocks with content and timestamps - next_cursor: Timestamp for pagination (if more results available) - total_matches: Number of references found in this batch Examples: get_page_references("GTD") get_page_references("Project Alpha", limit=20) get_page_references("Meeting Notes", limit=5, cursor=1640995200000) """ try: client = get_roam_client() result = client.get_page_references(page_name, limit, cursor) return json.dumps(result, indent=2) except Exception as e: print(f"Error getting page references: {e}", file=sys.stderr) return f"Error: {str(e)}"
• main.py:130-209 (helper)

  Core implementation logic in RoamResearchMCPServer class. Executes Datalog queries for page references with pagination support, processes nested block data, converts to markdown hierarchy, and structures the response.
  def get_page_references(self, page_name: str, limit: int = 10, cursor: Optional[int] = None) -> Dict[str, Any]: """Get references to a specific page with markdown content and child blocks""" # Build the query with time-based sorting and pagination if cursor: # Use cursor-based pagination for subsequent pages query = """[:find (pull ?ref [: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 {:block/children [:block/string :block/uid :edit/time]}]}]}]}]) ?time :in $ ?PAGE ?cursor-time :where [?page :node/title ?PAGE] [?ref :block/refs ?page] [?ref :edit/time ?time] [(< ?time ?cursor-time)] ]""" data = {"query": query, "args": [page_name, cursor]} else: # First page - no cursor query = """[:find (pull ?ref [: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 {:block/children [:block/string :block/uid :edit/time]}]}]}]}]) ?time :in $ ?PAGE :where [?page :node/title ?PAGE] [?ref :block/refs ?page] [?ref :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) and apply limit results = raw_result.get("result", []) sorted_results = sorted(results, key=lambda x: x[1], reverse=True) limited_results = sorted_results[:limit] # Transform the result to only include markdown content with children simplified_result = [] next_cursor = None for item in limited_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}) next_cursor = timestamp result = {"result": simplified_result} # Add next_cursor if there are more results if len(results) > limit: result["next_cursor"] = next_cursor return result
• main.py:68-83 (helper)

  Recursive helper to convert Roam block tree to indented markdown representation, preserving hierarchy.
  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:33-57 (helper)

  Utility method for making authenticated HTTP requests to Roam Research API.
  def _make_request( self, method: str, endpoint: str, data: Optional[Dict] = None ) -> Dict[str, Any]: """Make HTTP request to Roam Research API""" url = f"{self.base_url}{endpoint}" try: if method == "GET": response = requests.get(url, headers=self.headers) elif method == "POST": response = requests.post(url, headers=self.headers, json=data) else: raise ValueError(f"Unsupported HTTP method: {method}") response.raise_for_status() # Handle empty response for write operations if response.text.strip() == "": return {"result": "success", "status": response.status_code} return response.json() except requests.exceptions.RequestException as e: print(f"Request failed: {e}", file=sys.stderr) raise
• main.py:369-386 (helper)

  Singleton factory for initializing the RoamResearchMCPServer client instance from 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