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())