server.py

import asyncio
import json
import os
import logging
import httpx
from typing import Any, Dict, List, Optional
from dotenv import load_dotenv

from mcp.server.fastmcp import FastMCP

# Load environment variables from .env file
load_dotenv()

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("raindrop-mcp")

# Constants
API_URL = "https://api.raindrop.io/rest/v1"

# Create FastMCP server
mcp = FastMCP("Raindrop Collections API")

# Helper function to get headers with authentication token
async def get_headers():
    # Get token from .env file
    token = os.getenv("RAINDROP_TOKEN")
    if not token:
        logger.warning("RAINDROP_TOKEN not found in .env file.")
        logger.warning("Please add your Raindrop API token to the .env file.")
    
    # Use proper Bearer token format for authorization
    return {
        "Authorization": f"Bearer {token}" if token else "",
        "Content-Type": "application/json"
    }

# Define tools
@mcp.tool("get_root_collections")
async def get_root_collections() -> list:
    """
    Get all root collections from Raindrop.io
    """
    # Fetch collections from root endpoint
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        raise ValueError("API token not set. Please check your .env file.")
    
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{API_URL}/collections", headers=headers)
            
            if response.status_code != 200:
                raise ValueError(f"API returned status {response.status_code}")
            
            data = response.json()
            
            if "items" not in data:
                raise ValueError("Unexpected API response format")
            
            # Format response - return native Python list
            return [
                {
                    "_id": c.get("_id", ""),
                    "title": c.get("title", ""),
                    "count": c.get("count", 0),
                    "public": c.get("public", False),
                    "view": c.get("view", ""),
                    "color": c.get("color", ""),
                    "created": c.get("created", ""),
                    "lastUpdate": c.get("lastUpdate", ""),
                    "expanded": c.get("expanded", False)
                }
                for c in data["items"]
            ]
    except Exception as e:
        raise ValueError(f"Error fetching root collections: {str(e)}")

@mcp.tool("get_child_collections")
async def get_child_collections() -> str:
    """
    Get all child collections from Raindrop.io
    """
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        return json.dumps({
            "isError": True,
            "content": [{"type": "text", "text": "API token not set. Please check your .env file."}]
        })
    
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{API_URL}/collections/childrens", headers=headers)
            
            if response.status_code != 200:
                return json.dumps({
                    "isError": True,
                    "content": [{"type": "text", "text": f"Error: API returned status {response.status_code}"}]
                })
            
            data = response.json()
            
            if "items" not in data:
                return json.dumps({
                    "isError": True,
                    "content": [{"type": "text", "text": "Error: Unexpected API response format"}]
                })
            
            # Format response
            result = [
                {
                    "_id": c.get("_id", ""),
                    "title": c.get("title", ""),
                    "count": c.get("count", 0),
                    "public": c.get("public", False),
                    "view": c.get("view", ""),
                    "color": c.get("color", ""),
                    "parent_id": c.get("parent", {}).get("$id", None),
                    "created": c.get("created", ""),
                    "lastUpdate": c.get("lastUpdate", ""),
                    "expanded": c.get("expanded", False)
                }
                for c in data["items"]
            ]
            
            return json.dumps({
                "content": [
                    {"type": "text", "text": f"Found {len(result)} child collection(s)"},
                    {"type": "json", "json": result}
                ]
            })
    except Exception as e:
        return json.dumps({
            "isError": True,
            "content": [{"type": "text", "text": f"Error: {str(e)}"}]
        })

@mcp.tool("get_collection_by_id")
async def get_collection_by_id(collection_id: int) -> dict:
    """
    Get a specific collection from Raindrop.io by ID
    
    Args:
        collection_id: ID of the collection to fetch
    """
    # Validate input
    if collection_id is None:
        raise ValueError("No collection ID provided")
    
    # Fetch specific collection by ID
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        raise ValueError("API token not set. Please check your .env file.")
    
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{API_URL}/collection/{collection_id}", headers=headers)
            
            if response.status_code != 200:
                raise ValueError(f"API returned status {response.status_code}")
            
            data = response.json()
            
            if "item" not in data:
                raise ValueError("Unexpected API response format")
            
            # Format the single collection response
            collection = data["item"]
            result = {
                "_id": collection.get("_id", ""),
                "title": collection.get("title", ""),
                "count": collection.get("count", 0),
                "public": collection.get("public", False),
                "view": collection.get("view", ""),
                "color": collection.get("color", ""),
                "created": collection.get("created", ""),
                "lastUpdate": collection.get("lastUpdate", ""),
                "expanded": collection.get("expanded", False)
            }
            
            # Add parent ID if present
            if "parent" in collection and "$id" in collection["parent"]:
                result["parent_id"] = collection["parent"]["$id"]
            
            # Return native Python dict instead of JSON string
            return result
    except Exception as e:
        # Let exceptions bubble up to MCP
        raise ValueError(f"Error fetching collection: {str(e)}")

@mcp.tool("create_collection")
async def create_collection(
    title: str, 
    view: str = "list", 
    public: bool = False, 
    parent_id: Optional[int] = None
) -> str:
    """
    Create a new collection in Raindrop.io
    
    Args:
        title: Name of the collection
        view: View type (list, grid, masonry, simple)
        public: Whether the collection is public
        parent_id: ID of parent collection (omit for root collection)
    """
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        return json.dumps({
            "isError": True,
            "content": [{"type": "text", "text": "API token not set. Please check your .env file."}]
        })
        
    try:
        data = {
            "title": title,
            "view": view,
            "public": public
        }
        
        if parent_id is not None:
            data["parent"] = {"$id": parent_id}
            
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{API_URL}/collection",
                headers=headers,
                json=data
            )
            response_data = response.json()
            
            if not response_data.get("result", False):
                return json.dumps({
                    "isError": True,
                    "content": [{"type": "text", "text": f"Error creating collection: {response_data.get('errorMessage', 'Unknown error')}"}]
                })
            
            return json.dumps({
                "content": [
                    {"type": "text", "text": f"Collection '{title}' created successfully."},
                    {"type": "text", "text": f"Collection ID: {response_data.get('item', {}).get('_id', 'Unknown')}"}
                ]
            })
    except Exception as e:
        raise e  # Let exceptions bubble up
            
@mcp.tool("update_collection")
async def update_collection(
    collection_id: int, 
    title: Optional[str] = None,
    view: Optional[str] = None,
    public: Optional[bool] = None,
    parent_id: Optional[int] = None,
    expanded: Optional[bool] = None
) -> str:
    """
    Update an existing collection in Raindrop.io
    
    Args:
        collection_id: ID of the collection to update
        title: New name for the collection
        view: View type (list, grid, masonry, simple)
        public: Whether the collection is public
        parent_id: ID of parent collection (omit for root collection)
        expanded: Whether the collection is expanded
    """
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        return json.dumps({
            "isError": True,
            "content": [{"type": "text", "text": "API token not set. Please check your .env file."}]
        })
        
    try:
        data = {}
        if title is not None:
            data["title"] = title
        if view is not None:
            data["view"] = view
        if public is not None:
            data["public"] = public
        if parent_id is not None:
            data["parent"] = {"$id": parent_id}
        if expanded is not None:
            data["expanded"] = expanded
            
        if not data:
            return json.dumps({
                "isError": True,
                "content": [{"type": "text", "text": "No update parameters provided."}]
            })
            
        async with httpx.AsyncClient() as client:
            response = await client.put(
                f"{API_URL}/collection/{collection_id}",
                headers=headers,
                json=data
            )
            response_data = response.json()
            
            if not response_data.get("result", False):
                return json.dumps({
                    "isError": True,
                    "content": [{"type": "text", "text": f"Error updating collection: {response_data.get('errorMessage', 'Unknown error')}"}]
                })
            
            return json.dumps({
                "content": [
                    {"type": "text", "text": f"Collection {collection_id} updated successfully."}
                ]
            })
    except Exception as e:
        raise e  # Let exceptions bubble up
    
@mcp.tool("delete_collection")
async def delete_collection(collection_id: int) -> str:
    """
    Delete a collection from Raindrop.io. The raindrops will be moved to Trash.
    
    Args:
        collection_id: ID of the collection to delete
    """
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        return json.dumps({
            "isError": True,
            "content": [{"type": "text", "text": "API token not set. Please check your .env file."}]
        })
        
    try:
        async with httpx.AsyncClient() as client:
            response = await client.delete(
                f"{API_URL}/collection/{collection_id}",
                headers=headers
            )
            response_data = response.json()
            
            if not response_data.get("result", False):
                return json.dumps({
                    "isError": True,
                    "content": [{"type": "text", "text": f"Error deleting collection: {response_data.get('errorMessage', 'Unknown error')}"}]
                })
            
            return json.dumps({
                "content": [
                    {"type": "text", "text": f"Collection {collection_id} deleted successfully."}
                ]
            })
    except Exception as e:
        raise e  # Let exceptions bubble up
            
@mcp.tool("empty_trash")
async def empty_trash() -> str:
    """
    Empty the trash in Raindrop.io, permanently deleting all raindrops in it.
    """
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        return json.dumps({
            "isError": True,
            "content": [{"type": "text", "text": "API token not set. Please check your .env file."}]
        })
        
    try:
        async with httpx.AsyncClient() as client:
            response = await client.delete(
                f"{API_URL}/collection/-99",
                headers=headers
            )
            response_data = response.json()
            
            if not response_data.get("result", False):
                return json.dumps({
                    "isError": True,
                    "content": [{"type": "text", "text": f"Error emptying trash: {response_data.get('errorMessage', 'Unknown error')}"}]
                })
            
            return json.dumps({
                "content": [
                    {"type": "text", "text": "Trash emptied successfully."}
                ]
            })
    except Exception as e:
        raise e  # Let exceptions bubble up

@mcp.tool("get_raindrop")
async def get_raindrop(raindrop_id: int) -> dict:
    """
    Get a single raindrop from Raindrop.io by ID
    
    Args:
        raindrop_id: ID of the raindrop to fetch
    """
    # Validate input
    if raindrop_id is None:
        raise ValueError("No raindrop ID provided")
    
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        raise ValueError("API token not set. Please check your .env file.")
    
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(f"{API_URL}/raindrop/{raindrop_id}", headers=headers)
            
            if response.status_code != 200:
                raise ValueError(f"API returned status {response.status_code}")
            
            data = response.json()
            
            if "item" not in data:
                raise ValueError("Unexpected API response format")
            
            # Return the raindrop item directly
            return data["item"]
    except Exception as e:
        # Let exceptions bubble up to MCP
        raise ValueError(f"Error fetching raindrop: {str(e)}")

@mcp.tool("get_raindrops")
async def get_raindrops(
    collection_id: int,
    search: Optional[str] = None,
    sort: Optional[str] = None,
    page: Optional[int] = 0,
    perpage: Optional[int] = 25,
    nested: Optional[bool] = False
) -> dict:
    """
    Get multiple raindrops from a Raindrop.io collection
    
    Args:
        collection_id: ID of the collection to fetch raindrops from.
                       Use 0 for all raindrops, -1 for unsorted, -99 for trash.
        search: Optional search query
        sort: Sorting order. Options: -created (default), created, score, -sort, title, -title, domain, -domain
        page: Page number (starting from 0)
        perpage: Items per page (max 50)
        nested: Whether to include raindrops from nested collections
    """
    # Validate inputs
    if collection_id is None:
        raise ValueError("No collection ID provided")
    
    if perpage > 50:
        perpage = 50  # API limit
    
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        raise ValueError("API token not set. Please check your .env file.")
    
    try:
        # Build query parameters
        params = {}
        if search:
            params["search"] = search
        if sort:
            params["sort"] = sort
        if page is not None:
            params["page"] = page
        if perpage:
            params["perpage"] = perpage
        if nested:
            params["nested"] = "true"
        
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{API_URL}/raindrops/{collection_id}", 
                headers=headers,
                params=params
            )
            
            if response.status_code != 200:
                raise ValueError(f"API returned status {response.status_code}")
            
            return response.json()
    except Exception as e:
        # Let exceptions bubble up to MCP
        raise ValueError(f"Error fetching raindrops: {str(e)}")

@mcp.tool("get_tags")
async def get_tags(collection_id: Optional[int] = None) -> list:
    """
    Get tags from Raindrop.io
    
    Args:
        collection_id: Optional ID of the collection to fetch tags from.
                      When not specified, all tags from all collections will be retrieved.
    """
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        raise ValueError("API token not set. Please check your .env file.")
    
    try:
        # Build the endpoint URL
        endpoint = f"{API_URL}/tags"
        if collection_id is not None:
            endpoint = f"{endpoint}/{collection_id}"
        
        async with httpx.AsyncClient() as client:
            response = await client.get(endpoint, headers=headers)
            
            if response.status_code != 200:
                raise ValueError(f"API returned status {response.status_code}")
            
            data = response.json()
            
            if not data.get("result", False) or "items" not in data:
                raise ValueError("Unexpected API response format")
            
            # Return just the tags array for simplicity
            return data["items"]
    except Exception as e:
        # Let exceptions bubble up to MCP
        raise ValueError(f"Error fetching tags: {str(e)}")

@mcp.tool("update_raindrop")
async def update_raindrop(
    raindrop_id: int,
    title: Optional[str] = None,
    excerpt: Optional[str] = None,
    link: Optional[str] = None,
    important: Optional[bool] = None,
    tags: Optional[List[str]] = None,
    collection_id: Optional[int] = None,
    cover: Optional[str] = None,
    type: Optional[str] = None,
    order: Optional[int] = None,
    pleaseParse: Optional[bool] = None
) -> dict:
    """
    Update an existing raindrop (bookmark) in Raindrop.io
    
    Args:
        raindrop_id: ID of the raindrop to update
        title: New title for the raindrop
        excerpt: New description/excerpt
        link: New URL
        important: Set to True to mark as favorite
        tags: List of tags to assign
        collection_id: ID of collection to move the raindrop to
        cover: URL for the cover image
        type: Type of the raindrop
        order: Sort order (ascending) - set to 0 to move to first place
        pleaseParse: Set to True to reparse metadata (cover, type) in the background
    """
    # Validate input
    if raindrop_id is None:
        raise ValueError("No raindrop ID provided")
    
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        raise ValueError("API token not set. Please check your .env file.")
    
    try:
        # Build the request body with only provided parameters
        data = {}
        if title is not None:
            data["title"] = title
        if excerpt is not None:
            data["excerpt"] = excerpt
        if link is not None:
            data["link"] = link
        if important is not None:
            data["important"] = important
        if tags is not None:
            data["tags"] = tags
        if collection_id is not None:
            data["collection"] = {"$id": collection_id}
        if cover is not None:
            data["cover"] = cover
        if type is not None:
            data["type"] = type
        if order is not None:
            data["order"] = order
        if pleaseParse:
            data["pleaseParse"] = {}
        
        if not data:
            raise ValueError("No update parameters provided")
        
        async with httpx.AsyncClient() as client:
            response = await client.put(
                f"{API_URL}/raindrop/{raindrop_id}",
                headers=headers,
                json=data
            )
            
            if response.status_code != 200:
                raise ValueError(f"API returned status {response.status_code}")
            
            result = response.json()
            
            if not result.get("result", False):
                error_message = result.get("errorMessage", "Unknown error")
                raise ValueError(f"Error updating raindrop: {error_message}")
            
            return result
    except Exception as e:
        # Let exceptions bubble up to MCP
        raise ValueError(f"Error updating raindrop: {str(e)}")

@mcp.tool("update_many_raindrops")
async def update_many_raindrops(
    collection_id: int,
    ids: Optional[List[int]] = None,
    important: Optional[bool] = None,
    tags: Optional[List[str]] = None,
    cover: Optional[str] = None,
    target_collection_id: Optional[int] = None,
    nested: Optional[bool] = False,
    search: Optional[str] = None
) -> dict:
    """
    Update multiple raindrops at once within a collection
    
    Args:
        collection_id: ID of the collection containing raindrops to update
        ids: Optional list of specific raindrop IDs to update
        important: Set to True to mark as favorite, False to unmark
        tags: List of tags to add (or empty list to remove all tags)
        cover: URL for cover image (use '<screenshot>' to set screenshots for all)
        target_collection_id: ID of collection to move raindrops to
        nested: Include raindrops from nested collections
        search: Optional search query to filter which raindrops to update
    """
    # Validate input
    if collection_id is None:
        raise ValueError("No collection ID provided")
    
    headers = await get_headers()
    
    if not headers.get("Authorization"):
        raise ValueError("API token not set. Please check your .env file.")
    
    try:
        # Build the request body with only provided parameters
        data = {}
        if ids is not None:
            data["ids"] = ids
        if important is not None:
            data["important"] = important
        if tags is not None:
            data["tags"] = tags
        if cover is not None:
            data["cover"] = cover
        if target_collection_id is not None:
            data["collection"] = {"$id": target_collection_id}
        
        if not data:
            raise ValueError("No update parameters provided")
        
        # Build query params
        params = {}
        if search:
            params["search"] = search
        if nested:
            params["nested"] = "true"
        
        async with httpx.AsyncClient() as client:
            response = await client.put(
                f"{API_URL}/raindrops/{collection_id}",
                headers=headers,
                json=data,
                params=params
            )
            
            if response.status_code != 200:
                raise ValueError(f"API returned status {response.status_code}")
            
            result = response.json()
            
            if not result.get("result", False):
                error_message = result.get("errorMessage", "Unknown error")
                raise ValueError(f"Error updating raindrops: {error_message}")
            
            return result
    except Exception as e:
        # Let exceptions bubble up to MCP
        raise ValueError(f"Error updating raindrops: {str(e)}")

if __name__ == "__main__":
    asyncio.run(mcp.run())