claude-workspace/obsidian_mcp_server.py

import json, os, glob, shutil
from pathlib import Path
from datetime import datetime
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

server = Server("obsidian-mcp")
VAULT = Path(os.environ.get("OBSIDIAN_VAULT", "/Users/joungmin/Documents/Obsidian Vault"))

def note_path(relative: str) -> Path:
    """Resolve a relative note path inside the vault. Adds .md if missing."""
    p = VAULT / relative
    if not p.suffix:
        p = p.with_suffix(".md")
    return p

def rel(path: Path) -> str:
    return str(path.relative_to(VAULT))

@server.list_tools()
async def list_tools():
    return [
        types.Tool(name="list_folder", description="폴더 내 파일/폴더 목록 조회",
            inputSchema={"type": "object",
                "properties": {
                    "folder": {"type": "string", "description": "상대 경로 (비어두면 vault 루트)"}
                }}),
        types.Tool(name="search_notes", description="노트 내용 전체 텍스트 검색",
            inputSchema={"type": "object", "required": ["query"],
                "properties": {
                    "query": {"type": "string", "description": "검색 키워드"}
                }}),
        types.Tool(name="get_note", description="노트 내용 읽기",
            inputSchema={"type": "object", "required": ["path"],
                "properties": {
                    "path": {"type": "string", "description": "노트 상대 경로 (.md 생략 가능)"}
                }}),
        types.Tool(name="create_note", description="노트 생성 또는 덮어쓰기",
            inputSchema={"type": "object", "required": ["path", "content"],
                "properties": {
                    "path": {"type": "string"},
                    "content": {"type": "string"}
                }}),
        types.Tool(name="append_note", description="기존 노트 끝에 내용 추가",
            inputSchema={"type": "object", "required": ["path", "content"],
                "properties": {
                    "path": {"type": "string"},
                    "content": {"type": "string"}
                }}),
        types.Tool(name="delete_note", description="노트 삭제",
            inputSchema={"type": "object", "required": ["path"],
                "properties": {
                    "path": {"type": "string"}
                }}),
        types.Tool(name="move_note", description="노트 이동 또는 이름 변경",
            inputSchema={"type": "object", "required": ["src", "dst"],
                "properties": {
                    "src": {"type": "string", "description": "원본 상대 경로"},
                    "dst": {"type": "string", "description": "대상 상대 경로"}
                }}),
        types.Tool(name="create_daily_note", description="오늘의 Daily Note 생성 (이미 있으면 내용 반환)",
            inputSchema={"type": "object",
                "properties": {
                    "extra_content": {"type": "string", "description": "추가할 내용 (선택)"}
                }}),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):

    if name == "list_folder":
        folder = arguments.get("folder", "").strip("/")
        target = VAULT / folder if folder else VAULT
        if not target.exists():
            return [types.TextContent(type="text", text=f"폴더 없음: {folder}")]
        items = []
        for p in sorted(target.iterdir()):
            if p.name.startswith("."):
                continue
            items.append({
                "name": p.name,
                "type": "folder" if p.is_dir() else "file",
                "path": rel(p)
            })
        return [types.TextContent(type="text", text=json.dumps(items, ensure_ascii=False, indent=2))]

    elif name == "search_notes":
        query = arguments["query"].lower()
        results = []
        for md in VAULT.rglob("*.md"):
            try:
                text = md.read_text(encoding="utf-8", errors="ignore")
                if query in text.lower() or query in md.name.lower():
                    # Extract surrounding context
                    lines = text.splitlines()
                    matched = [l for l in lines if query in l.lower()]
                    results.append({
                        "path": rel(md),
                        "matches": matched[:3]
                    })
            except Exception:
                continue
        return [types.TextContent(type="text",
            text=json.dumps(results, ensure_ascii=False, indent=2)
                 if results else "검색 결과 없음")]

    elif name == "get_note":
        p = note_path(arguments["path"])
        if not p.exists():
            return [types.TextContent(type="text", text=f"노트 없음: {arguments['path']}")]
        return [types.TextContent(type="text", text=p.read_text(encoding="utf-8"))]

    elif name == "create_note":
        p = note_path(arguments["path"])
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(arguments["content"], encoding="utf-8")
        return [types.TextContent(type="text", text=f"저장됨: {rel(p)}")]

    elif name == "append_note":
        p = note_path(arguments["path"])
        if not p.exists():
            return [types.TextContent(type="text", text=f"노트 없음: {arguments['path']}")]
        with p.open("a", encoding="utf-8") as f:
            f.write("\n" + arguments["content"])
        return [types.TextContent(type="text", text=f"추가됨: {rel(p)}")]

    elif name == "delete_note":
        p = note_path(arguments["path"])
        if not p.exists():
            return [types.TextContent(type="text", text=f"노트 없음: {arguments['path']}")]
        p.unlink()
        return [types.TextContent(type="text", text=f"삭제됨: {rel(p)}")]

    elif name == "move_note":
        src = note_path(arguments["src"])
        dst = note_path(arguments["dst"])
        if not src.exists():
            return [types.TextContent(type="text", text=f"원본 없음: {arguments['src']}")]
        dst.parent.mkdir(parents=True, exist_ok=True)
        shutil.move(str(src), str(dst))
        return [types.TextContent(type="text", text=f"이동됨: {rel(src)} → {rel(dst)}")]

    elif name == "create_daily_note":
        today = datetime.now().strftime("%Y-%m-%d")
        p = VAULT / "Daily Notes" / f"{today}.md"
        p.parent.mkdir(parents=True, exist_ok=True)
        extra = arguments.get("extra_content", "")
        if p.exists():
            if extra:
                with p.open("a", encoding="utf-8") as f:
                    f.write("\n" + extra)
                return [types.TextContent(type="text", text=f"기존 노트에 추가됨: {rel(p)}")]
            return [types.TextContent(type="text", text=p.read_text(encoding="utf-8"))]
        content = f"# {today}\n\n{extra}" if extra else f"# {today}\n\n"
        p.write_text(content, encoding="utf-8")
        return [types.TextContent(type="text", text=f"생성됨: {rel(p)}")]

    return [types.TextContent(type="text", text=f"알 수 없는 도구: {name}")]

if __name__ == "__main__":
    import asyncio

    async def main():
        async with stdio_server() as (read_stream, write_stream):
            await server.run(read_stream, write_stream, server.create_initialization_options())

    asyncio.run(main())