Kiseki-Labs-Readwise-MCP

Instructions

Implementation Reference

• server.py:98-130 (handler)

  The main handler function for the 'get_readwise_highlights_by_document_ids' tool. It validates the input list of document IDs, creates concurrent tasks using the helper function for each ID, awaits the gathered results, and flattens them into a single list of highlights.
  @mcp.tool() async def get_readwise_highlights_by_document_ids( document_ids: List[int], ) -> List[Highlight]: """ Get highlights from Readwise by document ids. Args: document_ids (List[int]): The IDs of the documents to retrieve highlights for. Returns: List[Highlight]: A list of Highlight objects containing the highlights from the specified document. Raises: ValueError: If no document IDs are provided. """ if not document_ids: raise ValueError("No document IDs provided") # Create a list of tasks (co-routines), one for each document ID tasks = [get_highlight_by_document_id(READWISE_API_KEY, doc_id) for doc_id in document_ids] # Execute all tasks concurrently and gather the results results = await asyncio.gather(*tasks) highlights: List[Highlight] = [] # Flatten the list of lists into a single list of highlights for doc_highlights in results: highlights.extend(doc_highlights) return highlights
• readwise_mcp/types/highlight.py:13-27 (schema)

  Pydantic model used to validate and structure each highlight object returned by the tool.
  class Highlight(BaseModel): """Represents a highlight from Readwise API.""" id: int text: str note: str location: int location_type: str highlighted_at: Optional[datetime] = None url: Optional[HttpUrl] = None color: str updated: datetime book_id: int tags: List[Tag] = []
• readwise_mcp/tools/readwise/get_highlights.py:16-43 (helper)

  Supporting helper function that retrieves all highlights for a single document ID from the Readwise API, handling pagination across multiple requests.
  async def get_highlight_by_document_id(api_key: str, document_id: int) -> List[Highlight]: """Get highlights by document id.""" url = f"{READWISE_API_URL}/highlights/" params = {"book_id": document_id, "page_size": 100} highlights: List[Highlight] = [] first_request = True while True: # Pass params only on the first request. current_params = params if first_request else None hs = await get_data(api_key, url, current_params) first_request = False hs_results = hs["results"] highlights.extend([Highlight(**h) for h in hs_results]) total_highlights = hs["count"] logging.info(f"Total highlights: {total_highlights}") url = hs.get("next", None) if not url: break await asyncio.sleep(DEFAULT_SLEEP_BETWEEN_REQUESTS_IN_SECONDS) return highlights