NoteDiscovery/mcp_server_http/server.py

#!/usr/bin/env python3
"""
HTTP Streamable MCP Server for NoteDiscovery.

This server exposes NoteDiscovery tools over SSE using FastMCP.
"""

import sys
import os
import json
from typing import Optional

# Add parent directory to path so we can import from mcp_server
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field, AliasChoices

from mcp_server.config import load_config
from mcp_server.client import NoteDiscoveryClient

from mcp.server.transport_security import TransportSecuritySettings

# Initialize FastMCP server
mcp = FastMCP(
    "notediscovery_mcp_http",
    transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False)
)

# Load config and client
try:
    config = load_config()
    client = NoteDiscoveryClient(config)
except ValueError as e:
    print(f"Configuration error: {e}", file=sys.stderr)
    sys.exit(1)

# Helper for responses
def format_response(response) -> str:
    if not response.success:
        return f"Error: {response.error}"
    if response.data is None:
        return "Success (no data)"
    return json.dumps(response.data, indent=2, ensure_ascii=False)

# Pydantic models for tools
class SearchNotesInput(BaseModel):
    query: str = Field(..., description="Search query")
    limit: Optional[int] = Field(None, description="Maximum results to return")
    offset: int = Field(0, description="Pagination offset")

class ListNotesInput(BaseModel):
    limit: Optional[int] = Field(None, description="Maximum results to return")
    offset: int = Field(0, description="Pagination offset")

class PathInput(BaseModel):
    path: str = Field(..., validation_alias=AliasChoices('path', 'note', 'name'), description="Note or folder path")

class TagInput(BaseModel):
    tag: str = Field(..., description="Tag name")
    limit: Optional[int] = Field(None, description="Maximum results to return")
    offset: int = Field(0, description="Pagination offset")

class CreateNoteInput(BaseModel):
    path: str = Field(..., validation_alias=AliasChoices('path', 'note', 'name'), description="Note path")
    content: str = Field(..., description="Markdown content")

class AppendNoteInput(BaseModel):
    path: str = Field(..., validation_alias=AliasChoices('path', 'note', 'name'), description="Note path")
    content: str = Field(..., description="Content to append")
    add_timestamp: bool = Field(False, description="Whether to add timestamp")

class MoveNoteInput(BaseModel):
    old_path: str = Field(..., validation_alias=AliasChoices('old_path', 'old_name', 'old_note', 'source'), description="Current path")
    new_path: str = Field(..., validation_alias=AliasChoices('new_path', 'new_name', 'new_note', 'destination', 'target'), description="New path")

class CreateFromTemplateInput(BaseModel):
    template_name: str = Field(..., validation_alias=AliasChoices('template_name', 'template', 'name'), description="Template name")
    note_path: str = Field(..., validation_alias=AliasChoices('note_path', 'path', 'note', 'name'), description="Note path")

class GetTemplateInput(BaseModel):
    name: str = Field(..., description="Template name")

# Register Tools
@mcp.tool()
def search_notes(params: SearchNotesInput) -> str:
    """Search notes by query."""
    resp = client.search(params.query, limit=params.limit, offset=params.offset)
    return format_response(resp)

@mcp.tool()
def list_notes(params: ListNotesInput) -> str:
    """List all notes."""
    resp = client.list_notes(limit=params.limit, offset=params.offset)
    return format_response(resp)

@mcp.tool()
def get_note(params: PathInput) -> str:
    """Get note content."""
    resp = client.get_note(params.path)
    return format_response(resp)

@mcp.tool()
def list_tags() -> str:
    """List all tags."""
    resp = client.list_tags()
    return format_response(resp)

@mcp.tool()
def get_notes_by_tag(params: TagInput) -> str:
    """Get notes with a specific tag."""
    resp = client.get_notes_by_tag(params.tag, limit=params.limit, offset=params.offset)
    return format_response(resp)

@mcp.tool()
def get_graph() -> str:
    """Get knowledge graph data."""
    resp = client.get_graph()
    return format_response(resp)

@mcp.tool()
def get_backlinks(params: PathInput) -> str:
    """Get backlinks for a note."""
    resp = client.get_backlinks(params.path)
    return format_response(resp)

@mcp.tool()
def create_note(params: CreateNoteInput) -> str:
    """Create or update a note."""
    resp = client.create_note(params.path, params.content)
    return format_response(resp)

@mcp.tool()
def delete_note(params: PathInput) -> str:
    """Delete a note."""
    resp = client.delete_note(params.path)
    return format_response(resp)

@mcp.tool()
def create_folder(params: PathInput) -> str:
    """Create a folder."""
    resp = client.create_folder(params.path)
    return format_response(resp)

@mcp.tool()
def list_templates() -> str:
    """List templates."""
    resp = client.list_templates()
    return format_response(resp)

@mcp.tool()
def get_template(params: GetTemplateInput) -> str:
    """Get template content."""
    resp = client.get_template(params.name)
    return format_response(resp)

@mcp.tool()
def append_to_note(params: AppendNoteInput) -> str:
    """Append content to an existing note."""
    resp = client.append_to_note(params.path, params.content, params.add_timestamp)
    return format_response(resp)

@mcp.tool()
def move_note(params: MoveNoteInput) -> str:
    """Move or rename a note."""
    resp = client.move_note(params.old_path, params.new_path)
    return format_response(resp)

@mcp.tool()
def create_note_from_template(params: CreateFromTemplateInput) -> str:
    """Create a note from a template."""
    resp = client.create_note_from_template(params.template_name, params.note_path)
    return format_response(resp)

@mcp.tool()
def health_check() -> str:
    """Check server health."""
    resp = client.health_check()
    return format_response(resp)

@mcp.tool()
def get_config() -> str:
    """Get NoteDiscovery server configuration."""
    resp = client.get_config()
    return format_response(resp)

if __name__ == "__main__":
    port = int(os.environ.get("PORT", "8001"))
    host = os.environ.get("HOST", "0.0.0.0")
    print(f"Starting HTTP streamable MCP server on {host}:{port}", file=sys.stderr)
    
    app = mcp.streamable_http_app()
    from starlette.middleware.cors import CORSMiddleware
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    
    import uvicorn
    uvicorn.run(app, host=host, port=port)