server.py

Source file: 03-mcp/templates/python-server/server.py. Copy it from the repository, or use the copy button on the code block.

server.py

"""Template MCP server (MCP Python SDK v2 `MCPServer`, stdio) exposing search_notes and add_note.

Run:     uv run server.py            (or: python server.py)
Inspect: uv run mcp dev server.py
"""
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Annotated

from mcp.server.mcpserver import MCPServer
from mcp.types import ToolAnnotations
from pydantic import Field

NOTES_FILE = Path(os.environ.get("NOTES_FILE", str(Path.home() / ".notes-mcp.json")))

mcp = MCPServer("notes", version="0.1.0")


def log(msg: str) -> None:
    # stdout carries the MCP protocol. Log only to stderr.
    print(f"[notes] {msg}", file=sys.stderr)


def load_notes() -> list[dict]:
    if not NOTES_FILE.exists():
        return []
    return json.loads(NOTES_FILE.read_text(encoding="utf-8"))


def save_notes(notes: list[dict]) -> None:
    NOTES_FILE.write_text(json.dumps(notes, indent=2), encoding="utf-8")


def search(notes: list[dict], query: str, limit: int) -> list[dict]:
    q = query.lower()
    hits = [n for n in notes if q in f"{n['title']} {n['body']}".lower()]
    return sorted(hits, key=lambda n: n["created"], reverse=True)[:limit]


@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False))
def search_notes(
    query: Annotated[str, Field(min_length=1, description="Keyword or phrase, case-insensitive")],
    limit: Annotated[int, Field(ge=1, le=20, description="Maximum results (1-20)")] = 5,
) -> str:
    """Search saved notes by keyword in title or body. Returns up to `limit`
    matches, newest first. Use before add_note to avoid duplicates."""
    hits = search(load_notes(), query, limit)
    if not hits:
        return f'No notes match "{query}". Try a broader keyword.'
    return "\n".join(f"- {n['title']} ({n['created']}): {n['body'][:200]}" for n in hits)


@mcp.tool(annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, idempotentHint=False, openWorldHint=False))
def add_note(
    title: Annotated[str, Field(min_length=1, max_length=120, description="Unique note title")],
    body: Annotated[str, Field(min_length=1, max_length=10_000, description="Note content (Markdown allowed)")],
) -> str:
    """Save a new note. Fails if a note with the same title already exists."""
    notes = load_notes()
    if any(n["title"].lower() == title.lower() for n in notes):
        raise ValueError(f'A note titled "{title}" already exists. Choose a different title or use search_notes to find it.')
    notes.append({"title": title, "body": body, "created": datetime.now(timezone.utc).isoformat()})
    save_notes(notes)
    log(f'saved note "{title}"')
    return f'Saved note "{title}".'


@mcp.resource("notes://all", mime_type="application/json")
def all_notes() -> str:
    """Every saved note as JSON."""
    return json.dumps(load_notes(), indent=2)


@mcp.prompt()
def weekly_review(focus: str = "") -> str:
    """Summarise this week's notes."""
    suffix = f" focused on {focus}." if focus else "."
    return f"Search my notes from this week and write a weekly review{suffix}"


if __name__ == "__main__":
    log(f"running on stdio (NOTES_FILE={NOTES_FILE})")
    mcp.run()