From 1b4131cec04b57da1089c6537a1cec2be3a9e1ab Mon Sep 17 00:00:00 2001 From: Vincent Batts Date: Thu, 5 Jun 2025 16:08:11 -0400 Subject: [PATCH] claude.ai: > write a python program to run as a "tool" (https://openwebui.com/tools) for openweb-ui, which connects to a Readeck intance (https://codeberg.org/readeck/readeck). This tool exposes OpenAPI endpoints to get bookmarks and collections from the user's Readeck account. --- main.py | 260 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 260 insertions(+) create mode 100644 main.py diff --git a/main.py b/main.py new file mode 100644 index 0000000..7f85d08 --- /dev/null +++ b/main.py @@ -0,0 +1,260 @@ +""" +title: Readeck Integration +author: Assistant +author_url: https://github.com/assistant +funding_url: https://github.com/assistant +version: 0.1.0 +license: MIT +""" + +import requests +import json +from typing import Dict, List, Optional, Any +from pydantic import BaseModel, Field +import logging + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + +class Tools: + class Valves(BaseModel): + readeck_url: str = Field( + default="http://localhost:8000", + description="Readeck instance URL (e.g., http://localhost:8000 or https://readeck.example.com)" + ) + api_token: str = Field( + default="", + description="Readeck API token for authentication" + ) + + def __init__(self): + self.valves = self.Valves() + + def _get_headers(self) -> Dict[str, str]: + """Get headers for Readeck API requests""" + return { + "Authorization": f"Bearer {self.valves.api_token}", + "Content-Type": "application/json", + "Accept": "application/json" + } + + def _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Dict[str, Any]: + """Make a request to the Readeck API""" + url = f"{self.valves.readeck_url.rstrip('/')}/api{endpoint}" + + try: + response = requests.get( + url, + headers=self._get_headers(), + params=params or {}, + timeout=30 + ) + response.raise_for_status() + return response.json() + except requests.exceptions.RequestException as e: + logger.error(f"Error making request to {url}: {e}") + return {"error": str(e), "status_code": getattr(e.response, 'status_code', None)} + + async def get_bookmarks( + self, + limit: int = 20, + offset: int = 0, + collection_id: Optional[str] = None, + search: Optional[str] = None, + tags: Optional[str] = None, + __user__: Optional[dict] = None + ) -> str: + """ + Get bookmarks from Readeck + + Args: + limit: Number of bookmarks to return (default: 20, max: 100) + offset: Number of bookmarks to skip (default: 0) + collection_id: Filter by collection ID + search: Search query for bookmark titles/content + tags: Comma-separated list of tags to filter by + """ + + if not self.valves.readeck_url or not self.valves.api_token: + return "Error: Readeck URL and API token must be configured in the tool settings." + + # Build query parameters + params = { + "limit": min(limit, 100), # Cap at 100 to prevent excessive requests + "offset": offset + } + + if collection_id: + params["collection_id"] = collection_id + if search: + params["q"] = search + if tags: + params["tags"] = tags + + # Make request to Readeck API + result = self._make_request("/bookmarks", params) + + if "error" in result: + return f"Error fetching bookmarks: {result['error']}" + + # Format the response + bookmarks = result.get("bookmarks", []) + total = result.get("total", len(bookmarks)) + + if not bookmarks: + return "No bookmarks found matching your criteria." + + # Format bookmarks for display + formatted_bookmarks = [] + for bookmark in bookmarks: + formatted_bookmark = { + "id": bookmark.get("id"), + "title": bookmark.get("title", "Untitled"), + "url": bookmark.get("url"), + "excerpt": bookmark.get("excerpt", "")[:200] + "..." if bookmark.get("excerpt", "") else "", + "tags": bookmark.get("tags", []), + "collection": bookmark.get("collection", {}).get("name", ""), + "created_at": bookmark.get("created_at"), + "updated_at": bookmark.get("updated_at") + } + formatted_bookmarks.append(formatted_bookmark) + + return json.dumps({ + "bookmarks": formatted_bookmarks, + "total": total, + "limit": params["limit"], + "offset": params["offset"] + }, indent=2) + + async def get_collections( + self, + __user__: Optional[dict] = None + ) -> str: + """ + Get collections from Readeck + """ + + if not self.valves.readeck_url or not self.valves.api_token: + return "Error: Readeck URL and API token must be configured in the tool settings." + + # Make request to Readeck API + result = self._make_request("/collections") + + if "error" in result: + return f"Error fetching collections: {result['error']}" + + collections = result.get("collections", []) + + if not collections: + return "No collections found." + + # Format collections for display + formatted_collections = [] + for collection in collections: + formatted_collection = { + "id": collection.get("id"), + "name": collection.get("name"), + "description": collection.get("description", ""), + "bookmark_count": collection.get("bookmark_count", 0), + "created_at": collection.get("created_at"), + "updated_at": collection.get("updated_at") + } + formatted_collections.append(formatted_collection) + + return json.dumps({ + "collections": formatted_collections, + "total": len(formatted_collections) + }, indent=2) + + async def get_bookmark_by_id( + self, + bookmark_id: str, + __user__: Optional[dict] = None + ) -> str: + """ + Get a specific bookmark by its ID + + Args: + bookmark_id: The ID of the bookmark to retrieve + """ + + if not self.valves.readeck_url or not self.valves.api_token: + return "Error: Readeck URL and API token must be configured in the tool settings." + + if not bookmark_id: + return "Error: bookmark_id is required." + + # Make request to Readeck API + result = self._make_request(f"/bookmarks/{bookmark_id}") + + if "error" in result: + return f"Error fetching bookmark: {result['error']}" + + bookmark = result.get("bookmark") + if not bookmark: + return f"Bookmark with ID {bookmark_id} not found." + + # Format bookmark for display + formatted_bookmark = { + "id": bookmark.get("id"), + "title": bookmark.get("title", "Untitled"), + "url": bookmark.get("url"), + "content": bookmark.get("content", ""), + "excerpt": bookmark.get("excerpt", ""), + "tags": bookmark.get("tags", []), + "collection": bookmark.get("collection", {}), + "created_at": bookmark.get("created_at"), + "updated_at": bookmark.get("updated_at"), + "reading_time": bookmark.get("reading_time"), + "word_count": bookmark.get("word_count") + } + + return json.dumps(formatted_bookmark, indent=2) + + async def search_bookmarks( + self, + query: str, + limit: int = 10, + __user__: Optional[dict] = None + ) -> str: + """ + Search bookmarks by title, content, or URL + + Args: + query: Search query + limit: Number of results to return (default: 10) + """ + + if not query: + return "Error: search query is required." + + return await self.get_bookmarks( + limit=limit, + search=query, + __user__=__user__ + ) + + async def get_bookmarks_by_tag( + self, + tags: str, + limit: int = 20, + __user__: Optional[dict] = None + ) -> str: + """ + Get bookmarks filtered by tags + + Args: + tags: Comma-separated list of tags + limit: Number of results to return (default: 20) + """ + + if not tags: + return "Error: tags parameter is required." + + return await self.get_bookmarks( + limit=limit, + tags=tags, + __user__=__user__ + ) \ No newline at end of file