#!/usr/bin/env python3
"""
MCP Server để quản lý ghi chú cá nhân
Cung cấp các công cụ để tạo, đọc, cập nhật, xóa và tìm kiếm ghi chú
"""
import json
import os
from datetime import datetime
from pathlib import Path
from typing import Any
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
# Thư mục lưu trữ ghi chú
NOTES_DIR = Path.home() / ".kiro_notes"
NOTES_DIR.mkdir(exist_ok=True)
# Khởi tạo server
server = Server("notes-manager")
def get_note_path(note_id: str) -> Path:
"""Lấy đường dẫn file ghi chú"""
return NOTES_DIR / f"{note_id}.json"
def list_all_notes() -> list[dict[str, Any]]:
"""Liệt kê tất cả ghi chú"""
notes = []
for file_path in NOTES_DIR.glob("*.json"):
try:
with open(file_path, "r", encoding="utf-8") as f:
note = json.load(f)
notes.append(note)
except Exception:
continue
return sorted(notes, key=lambda x: x.get("updated_at", ""), reverse=True)
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""Danh sách các công cụ có sẵn"""
return [
types.Tool(
name="create_note",
description="Tạo ghi chú mới với tiêu đề và nội dung",
inputSchema={
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "Tiêu đề ghi chú"
},
"content": {
"type": "string",
"description": "Nội dung ghi chú"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Các thẻ tag (tùy chọn)"
}
},
"required": ["title", "content"]
}
),
types.Tool(
name="read_note",
description="Đọc nội dung ghi chú theo ID",
inputSchema={
"type": "object",
"properties": {
"note_id": {
"type": "string",
"description": "ID của ghi chú"
}
},
"required": ["note_id"]
}
),
types.Tool(
name="update_note",
description="Cập nhật ghi chú hiện có",
inputSchema={
"type": "object",
"properties": {
"note_id": {
"type": "string",
"description": "ID của ghi chú"
},
"title": {
"type": "string",
"description": "Tiêu đề mới (tùy chọn)"
},
"content": {
"type": "string",
"description": "Nội dung mới (tùy chọn)"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Các thẻ tag mới (tùy chọn)"
}
},
"required": ["note_id"]
}
),
types.Tool(
name="delete_note",
description="Xóa ghi chú theo ID",
inputSchema={
"type": "object",
"properties": {
"note_id": {
"type": "string",
"description": "ID của ghi chú cần xóa"
}
},
"required": ["note_id"]
}
),
types.Tool(
name="list_notes",
description="Liệt kê tất cả ghi chú với thông tin tóm tắt",
inputSchema={
"type": "object",
"properties": {
"tag": {
"type": "string",
"description": "Lọc theo tag (tùy chọn)"
}
}
}
),
types.Tool(
name="search_notes",
description="Tìm kiếm ghi chú theo từ khóa trong tiêu đề hoặc nội dung",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Từ khóa tìm kiếm"
}
},
"required": ["query"]
}
)
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""Xử lý các lệnh gọi công cụ"""
if name == "create_note":
title = arguments.get("title", "")
content = arguments.get("content", "")
tags = arguments.get("tags", [])
# Tạo ID duy nhất
note_id = datetime.now().strftime("%Y%m%d_%H%M%S")
timestamp = datetime.now().isoformat()
note = {
"id": note_id,
"title": title,
"content": content,
"tags": tags,
"created_at": timestamp,
"updated_at": timestamp
}
# Lưu ghi chú
note_path = get_note_path(note_id)
with open(note_path, "w", encoding="utf-8") as f:
json.dump(note, f, ensure_ascii=False, indent=2)
return [
types.TextContent(
type="text",
text=f"✅ Đã tạo ghi chú thành công!\nID: {note_id}\nTiêu đề: {title}"
)
]
elif name == "read_note":
note_id = arguments.get("note_id", "")
note_path = get_note_path(note_id)
if not note_path.exists():
return [types.TextContent(type="text", text=f"❌ Không tìm thấy ghi chú với ID: {note_id}")]
with open(note_path, "r", encoding="utf-8") as f:
note = json.load(f)
result = f"""📝 Ghi chú: {note['title']}
ID: {note['id']}
Tạo lúc: {note['created_at']}
Cập nhật: {note['updated_at']}
Tags: {', '.join(note.get('tags', []))}
Nội dung:
{note['content']}
"""
return [types.TextContent(type="text", text=result)]
elif name == "update_note":
note_id = arguments.get("note_id", "")
note_path = get_note_path(note_id)
if not note_path.exists():
return [types.TextContent(type="text", text=f"❌ Không tìm thấy ghi chú với ID: {note_id}")]
with open(note_path, "r", encoding="utf-8") as f:
note = json.load(f)
# Cập nhật các trường
if "title" in arguments:
note["title"] = arguments["title"]
if "content" in arguments:
note["content"] = arguments["content"]
if "tags" in arguments:
note["tags"] = arguments["tags"]
note["updated_at"] = datetime.now().isoformat()
with open(note_path, "w", encoding="utf-8") as f:
json.dump(note, f, ensure_ascii=False, indent=2)
return [types.TextContent(type="text", text=f"✅ Đã cập nhật ghi chú: {note['title']}")]
elif name == "delete_note":
note_id = arguments.get("note_id", "")
note_path = get_note_path(note_id)
if not note_path.exists():
return [types.TextContent(type="text", text=f"❌ Không tìm thấy ghi chú với ID: {note_id}")]
with open(note_path, "r", encoding="utf-8") as f:
note = json.load(f)
os.remove(note_path)
return [types.TextContent(type="text", text=f"🗑️ Đã xóa ghi chú: {note['title']}")]
elif name == "list_notes":
notes = list_all_notes()
tag_filter = arguments.get("tag") if arguments else None
if tag_filter:
notes = [n for n in notes if tag_filter in n.get("tags", [])]
if not notes:
return [types.TextContent(type="text", text="📭 Chưa có ghi chú nào")]
result = f"📚 Danh sách ghi chú ({len(notes)} ghi chú):\n\n"
for note in notes:
tags_str = f" [{', '.join(note.get('tags', []))}]" if note.get('tags') else ""
result += f"• {note['id']} - {note['title']}{tags_str}\n"
result += f" Cập nhật: {note['updated_at']}\n\n"
return [types.TextContent(type="text", text=result)]
elif name == "search_notes":
query = arguments.get("query", "").lower()
notes = list_all_notes()
# Tìm kiếm trong tiêu đề và nội dung
results = [
n for n in notes
if query in n.get("title", "").lower() or query in n.get("content", "").lower()
]
if not results:
return [types.TextContent(type="text", text=f"🔍 Không tìm thấy ghi chú nào với từ khóa: {query}")]
result = f"🔍 Tìm thấy {len(results)} ghi chú:\n\n"
for note in results:
result += f"• {note['id']} - {note['title']}\n"
# Hiển thị đoạn trích
content_preview = note['content'][:100] + "..." if len(note['content']) > 100 else note['content']
result += f" {content_preview}\n\n"
return [types.TextContent(type="text", text=result)]
else:
raise ValueError(f"Unknown tool: {name}")
async def main():
"""Chạy server"""
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="notes-manager",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
import asyncio
asyncio.run(main())