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.
This commit is contained in:
Vincent Batts 2025-06-05 16:08:11 -04:00
commit 1b4131cec0

260
main.py Normal file
View file

@ -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__
)