claude-configs/skills/mcp-manager/notediscovery_mcp.py

#!/usr/bin/env python3
"""
NoteDiscovery MCP Server
Full CRUD MCP server for NoteDiscovery knowledge base integration with Claude Code.

Tools:
  - search_notes: Full-text search across all notes
  - read_note: Get content of a specific note
  - create_note: Create a new note
  - update_note: Update an existing note
  - delete_note: Delete a note
  - list_notes: Browse folder structure
  - list_tags: Get all tags with counts
  - get_notes_by_tag: Get notes with a specific tag
  - get_graph: Get note relationship graph
"""

import json
import logging
import os
import sys
from typing import Any

import httpx

# MCP SDK imports
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# Configuration from environment
NOTEDISCOVERY_URL = os.environ.get("NOTEDISCOVERY_URL", "https://notes.manticorum.com")
NOTEDISCOVERY_PASSWORD = os.environ.get("NOTEDISCOVERY_PASSWORD", "")

# Set up logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger("notediscovery-mcp")

# Initialize MCP server
server = Server("notediscovery")


class NoteDiscoveryClient:
    """HTTP client for NoteDiscovery API with session-based auth."""

    def __init__(self, base_url: str, password: str):
        self.base_url = base_url.rstrip("/")
        self.password = password
        self.client = httpx.Client(timeout=30.0, follow_redirects=True)
        self._authenticated = False

    def _ensure_auth(self):
        """Authenticate if not already done."""
        if self._authenticated:
            return

        if not self.password:
            logger.warning("No password configured - requests may fail")
            return

        # Login to get session cookie
        response = self.client.post(
            f"{self.base_url}/login",
            data={"password": self.password},
            follow_redirects=False
        )

        if response.status_code in (200, 302, 303):
            self._authenticated = True
            logger.info("Successfully authenticated to NoteDiscovery")
        else:
            logger.error(f"Authentication failed: {response.status_code}")

    def _request(self, method: str, endpoint: str, **kwargs) -> dict | list | str:
        """Make authenticated request to API."""
        self._ensure_auth()

        url = f"{self.base_url}{endpoint}"
        response = self.client.request(method, url, **kwargs)

        if response.status_code == 401:
            # Try re-authenticating
            self._authenticated = False
            self._ensure_auth()
            response = self.client.request(method, url, **kwargs)

        if response.status_code >= 400:
            return {"error": f"HTTP {response.status_code}: {response.text}"}

        # Try JSON first, fall back to text
        try:
            return response.json()
        except json.JSONDecodeError:
            return response.text

    def search(self, query: str) -> dict:
        """Search notes by keyword."""
        return self._request("GET", "/api/search", params={"q": query})

    def list_notes(self) -> dict:
        """Get all notes with metadata."""
        return self._request("GET", "/api/notes")

    def read_note(self, path: str) -> dict:
        """Get content of a specific note."""
        return self._request("GET", f"/api/notes/{path}")

    def create_note(self, path: str, content: str) -> dict:
        """Create a new note."""
        return self._request("POST", f"/api/notes/{path}", json={"content": content})

    def update_note(self, path: str, content: str) -> dict:
        """Update an existing note."""
        return self._request("POST", f"/api/notes/{path}", json={"content": content})

    def delete_note(self, path: str) -> dict:
        """Delete a note."""
        return self._request("DELETE", f"/api/notes/{path}")

    def list_tags(self) -> dict:
        """Get all tags with counts."""
        return self._request("GET", "/api/tags")

    def get_notes_by_tag(self, tag: str) -> dict:
        """Get notes with a specific tag."""
        return self._request("GET", f"/api/tags/{tag}")

    def get_graph(self) -> dict:
        """Get note relationship graph."""
        return self._request("GET", "/api/graph")

    def create_folder(self, path: str) -> dict:
        """Create a new folder."""
        return self._request("POST", "/api/folders", json={"path": path})


# Initialize client
client = NoteDiscoveryClient(NOTEDISCOVERY_URL, NOTEDISCOVERY_PASSWORD)


@server.list_tools()
async def list_tools() -> list[Tool]:
    """List available tools."""
    return [
        Tool(
            name="search_notes",
            description="Search across all notes in the knowledge base. Returns matching notes with snippets.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query - searches note titles and content"
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="read_note",
            description="Read the full content of a specific note by its path.",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path to the note (e.g., 'folder/note-name' or 'note-name'). Do not include .md extension."
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="create_note",
            description="Create a new note in the knowledge base. Use markdown formatting.",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path for the new note (e.g., 'folder/note-name'). Creates folders if needed."
                    },
                    "content": {
                        "type": "string",
                        "description": "Markdown content for the note. Include a # heading as the title."
                    }
                },
                "required": ["path", "content"]
            }
        ),
        Tool(
            name="update_note",
            description="Update an existing note's content. Replaces the entire note content.",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path to the existing note"
                    },
                    "content": {
                        "type": "string",
                        "description": "New markdown content (replaces existing content)"
                    }
                },
                "required": ["path", "content"]
            }
        ),
        Tool(
            name="delete_note",
            description="Delete a note from the knowledge base. Use with caution.",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path to the note to delete"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="list_notes",
            description="List all notes and folders in the knowledge base. Returns hierarchical structure.",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        ),
        Tool(
            name="list_tags",
            description="Get all tags used in the knowledge base with note counts.",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        ),
        Tool(
            name="get_notes_by_tag",
            description="Get all notes that have a specific tag.",
            inputSchema={
                "type": "object",
                "properties": {
                    "tag": {
                        "type": "string",
                        "description": "Tag name (without #)"
                    }
                },
                "required": ["tag"]
            }
        ),
        Tool(
            name="get_graph",
            description="Get the relationship graph showing how notes link to each other.",
            inputSchema={
                "type": "object",
                "properties": {},
                "required": []
            }
        )
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    """Execute a tool and return results."""
    try:
        if name == "search_notes":
            result = client.search(arguments["query"])
        elif name == "read_note":
            result = client.read_note(arguments["path"])
        elif name == "create_note":
            result = client.create_note(arguments["path"], arguments["content"])
        elif name == "update_note":
            result = client.update_note(arguments["path"], arguments["content"])
        elif name == "delete_note":
            result = client.delete_note(arguments["path"])
        elif name == "list_notes":
            result = client.list_notes()
        elif name == "list_tags":
            result = client.list_tags()
        elif name == "get_notes_by_tag":
            result = client.get_notes_by_tag(arguments["tag"])
        elif name == "get_graph":
            result = client.get_graph()
        else:
            result = {"error": f"Unknown tool: {name}"}

        # Format result
        if isinstance(result, dict) or isinstance(result, list):
            text = json.dumps(result, indent=2)
        else:
            text = str(result)

        return [TextContent(type="text", text=text)]

    except Exception as e:
        logger.error(f"Error executing {name}: {e}")
        return [TextContent(type="text", text=json.dumps({"error": str(e)}))]


async def main():
    """Run the MCP server."""
    logger.info(f"Starting NoteDiscovery MCP server for {NOTEDISCOVERY_URL}")
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())