MCP Notes Connector

""" Evernote MCP Server - メインサーバー実装 MCPプロトコルの実装とEvernote APIとの統合を担当します。 """ import asyncio import os from typing import Any from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent from .evernote_client import EvernoteClient class EvernoteMCPServer: """Evernote MCP Server implementation.""" def __init__(self, token: str, sandbox: bool = False): """ Initialize the Evernote MCP Server. Args: token: Evernote API token sandbox: Whether to use sandbox environment """ self.server = Server("evernote-mcp-server") self.evernote_client = EvernoteClient(token, sandbox) self._setup_handlers() def _setup_handlers(self) -> None: """Set up MCP request handlers.""" @self.server.list_tools() async def list_tools() -> list[Tool]: """Return list of available tools.""" return [ Tool( name="list_notebooks", description="ユーザーのノートブック一覧を取得", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="search_notes", description="条件に基づいてノートを検索", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "検索クエリ（Evernote検索構文）", }, "max_results": { "type": "number", "description": "最大結果数（デフォルト: 50）", }, }, "required": ["query"], }, ), Tool( name="get_note", description="特定のノートの内容を取得", inputSchema={ "type": "object", "properties": { "guid": { "type": "string", "description": "ノートのGUID", }, }, "required": ["guid"], }, ), Tool( name="create_note", description="新しいノートを作成", inputSchema={ "type": "object", "properties": { "title": { "type": "string", "description": "ノートのタイトル", }, "content": { "type": "string", "description": "ノートの内容（ENML形式）", }, "notebook_guid": { "type": "string", "description": "ノートブックのGUID（オプション）", }, "tags": { "type": "array", "items": {"type": "string"}, "description": "タグのリスト", }, }, "required": ["title", "content"], }, ), ] @self.server.call_tool() async def call_tool(name: str, arguments: Any) -> list[TextContent]: """Handle tool calls.""" try: if name == "list_notebooks": result = await self._handle_list_notebooks() elif name == "search_notes": result = await self._handle_search_notes(arguments) elif name == "get_note": result = await self._handle_get_note(arguments) elif name == "create_note": result = await self._handle_create_note(arguments) else: raise ValueError(f"Unknown tool: {name}") return [TextContent(type="text", text=result)] except Exception as e: error_msg = f"Error: {str(e)}" return [TextContent(type="text", text=error_msg)] async def _handle_list_notebooks(self) -> str: """Handle list_notebooks tool call.""" try: notebooks = await self.evernote_client.list_notebooks() if not notebooks: return "ノートブックが見つかりませんでした。" # レスポンスを整形 result_lines = [f"📚 ノートブック一覧 (全{len(notebooks)}件)\n"] for nb in notebooks: name = nb["name"] guid = nb["guid"] stack = f" (スタック: {nb['stack']})" if nb.get("stack") else "" default = " ⭐デフォルト" if nb.get("default_notebook") else "" result_lines.append(f"• {name}{stack}{default}") result_lines.append(f" GUID: {guid}") return "\n".join(result_lines) except Exception as e: return f"ノートブック一覧の取得に失敗しました: {str(e)}" async def _handle_search_notes(self, arguments: dict[str, Any]) -> str: """Handle search_notes tool call.""" # TODO: Evernote APIでノートを検索 query = arguments.get("query") max_results = arguments.get("max_results", 50) return f"Not yet implemented (query={query}, max_results={max_results})" async def _handle_get_note(self, arguments: dict[str, Any]) -> str: """Handle get_note tool call.""" # TODO: Evernote APIでノートを取得 guid = arguments.get("guid") return f"Not yet implemented (guid={guid})" async def _handle_create_note(self, arguments: dict[str, Any]) -> str: """Handle create_note tool call.""" # TODO: Evernote APIで新しいノートを作成 title = arguments.get("title") content = arguments.get("content") notebook_guid = arguments.get("notebook_guid") tags = arguments.get("tags", []) return f"Not yet implemented (title={title})" async def run(self) -> None: """Start the MCP server.""" async with stdio_server() as (read_stream, write_stream): await self.server.run( read_stream, write_stream, self.server.create_initialization_options(), ) def main() -> None: """Main entry point.""" # 環境変数から設定を取得 from dotenv import load_dotenv load_dotenv() token = os.getenv("EVERNOTE_TOKEN") sandbox = os.getenv("EVERNOTE_SANDBOX", "false").lower() == "true" if not token: print("Error: EVERNOTE_TOKEN environment variable is required") exit(1) # サーバーを起動 server = EvernoteMCPServer(token, sandbox) asyncio.run(server.run()) if __name__ == "__main__": main()