Updating openapi and creating periodic tools

src/mcp_obsidian/obsidian.py

@@ -152,4 +152,88 @@ class Obsidian():
             response.raise_for_status()
             return response.json()
 
-        return self._safe_call(call_fn)
+        return self._safe_call(call_fn)
+    
+    def get_periodic_note(self, period: str) -> Any:
+        """Get current periodic note for the specified period.
+        
+        Args:
+            period: The period type (daily, weekly, monthly, quarterly, yearly)
+            
+        Returns:
+            Content of the periodic note
+        """
+        url = f"{self.get_base_url()}/periodic/{period}/"
+        
+        def call_fn():
+            response = requests.get(url, headers=self._get_headers(), verify=self.verify_ssl, timeout=self.timeout)
+            response.raise_for_status()
+            
+            return response.text
+
+        return self._safe_call(call_fn)
+    
+    def get_recent_periodic_notes(self, period: str, limit: int = 5, include_content: bool = False) -> Any:
+        """Get most recent periodic notes for the specified period type.
+        
+        Args:
+            period: The period type (daily, weekly, monthly, quarterly, yearly)
+            limit: Maximum number of notes to return (default: 5)
+            include_content: Whether to include note content (default: False)
+            
+        Returns:
+            List of recent periodic notes
+        """
+        url = f"{self.get_base_url()}/periodic/{period}/recent"
+        params = {
+            "limit": limit,
+            "includeContent": include_content
+        }
+        
+        def call_fn():
+            response = requests.get(
+                url, 
+                headers=self._get_headers(), 
+                params=params,
+                verify=self.verify_ssl, 
+                timeout=self.timeout
+            )
+            response.raise_for_status()
+            
+            return response.json()
+
+        return self._safe_call(call_fn)
+    
+    def get_recent_changes(self, limit: int = 10, days: int = None, extensions: list = None) -> Any:
+        """Get recently modified files in the vault.
+        
+        Args:
+            limit: Maximum number of files to return (default: 10)
+            days: Only include files modified within this many days (optional)
+            extensions: Only include files with these extensions (optional)
+            
+        Returns:
+            List of recently modified files with metadata
+        """
+        url = f"{self.get_base_url()}/vault/recent"
+        params = {"limit": limit}
+        
+        if days is not None:
+            params["days"] = days
+            
+        if extensions is not None:
+            params["extensions"] = ",".join(extensions)
+        
+        def call_fn():
+            response = requests.get(
+                url, 
+                headers=self._get_headers(), 
+                params=params,
+                verify=self.verify_ssl, 
+                timeout=self.timeout
+            )
+            response.raise_for_status()
+            
+            return response.json()
+
+        return self._safe_call(call_fn)

src/mcp_obsidian/server.py

@@ -49,6 +49,9 @@ add_tool_handler(tools.PatchContentToolHandler())
 add_tool_handler(tools.AppendContentToolHandler())
 add_tool_handler(tools.ComplexSearchToolHandler())
 add_tool_handler(tools.BatchGetFileContentsToolHandler())
+add_tool_handler(tools.PeriodicNotesToolHandler())
+add_tool_handler(tools.RecentPeriodicNotesToolHandler())
+add_tool_handler(tools.RecentChangesToolHandler())
 
 @app.list_tools()
 async def list_tools() -> list[Tool]:
@@ -86,6 +89,3 @@ async def main():
             write_stream,
             app.create_initialization_options()
         )
-
-
-

src/mcp_obsidian/tools.py

@@ -362,4 +362,161 @@ class BatchGetFileContentsToolHandler(ToolHandler):
                 type="text",
                 text=content
             )
-        ]
+        ]
+
+class PeriodicNotesToolHandler(ToolHandler):
+    def __init__(self):
+        super().__init__("obsidian_get_periodic_note")
+
+    def get_tool_description(self):
+        return Tool(
+            name=self.name,
+            description="Get current periodic note for the specified period.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "period": {
+                        "type": "string",
+                        "description": "The period type (daily, weekly, monthly, quarterly, yearly)",
+                        "enum": ["daily", "weekly", "monthly", "quarterly", "yearly"]
+                    }
+                },
+                "required": ["period"]
+            }
+        )
+
+    def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+        if "period" not in args:
+            raise RuntimeError("period argument missing in arguments")
+
+        period = args["period"]
+        valid_periods = ["daily", "weekly", "monthly", "quarterly", "yearly"]
+        if period not in valid_periods:
+            raise RuntimeError(f"Invalid period: {period}. Must be one of: {', '.join(valid_periods)}")
+
+        api = obsidian.Obsidian(api_key=api_key)
+        content = api.get_periodic_note(period)
+
+        return [
+            TextContent(
+                type="text",
+                text=content
+            )
+        ]
+        
+class RecentPeriodicNotesToolHandler(ToolHandler):
+    def __init__(self):
+        super().__init__("obsidian_get_recent_periodic_notes")
+
+    def get_tool_description(self):
+        return Tool(
+            name=self.name,
+            description="Get most recent periodic notes for the specified period type.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "period": {
+                        "type": "string",
+                        "description": "The period type (daily, weekly, monthly, quarterly, yearly)",
+                        "enum": ["daily", "weekly", "monthly", "quarterly", "yearly"]
+                    },
+                    "limit": {
+                        "type": "integer",
+                        "description": "Maximum number of notes to return (default: 5)",
+                        "default": 5,
+                        "minimum": 1,
+                        "maximum": 50
+                    },
+                    "include_content": {
+                        "type": "boolean",
+                        "description": "Whether to include note content (default: false)",
+                        "default": False
+                    }
+                },
+                "required": ["period"]
+            }
+        )
+
+    def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+        if "period" not in args:
+            raise RuntimeError("period argument missing in arguments")
+
+        period = args["period"]
+        valid_periods = ["daily", "weekly", "monthly", "quarterly", "yearly"]
+        if period not in valid_periods:
+            raise RuntimeError(f"Invalid period: {period}. Must be one of: {', '.join(valid_periods)}")
+
+        limit = args.get("limit", 5)
+        if not isinstance(limit, int) or limit < 1:
+            raise RuntimeError(f"Invalid limit: {limit}. Must be a positive integer")
+            
+        include_content = args.get("include_content", False)
+        if not isinstance(include_content, bool):
+            raise RuntimeError(f"Invalid include_content: {include_content}. Must be a boolean")
+
+        api = obsidian.Obsidian(api_key=api_key)
+        results = api.get_recent_periodic_notes(period, limit, include_content)
+
+        return [
+            TextContent(
+                type="text",
+                text=json.dumps(results, indent=2)
+            )
+        ]
+        
+class RecentChangesToolHandler(ToolHandler):
+    def __init__(self):
+        super().__init__("obsidian_get_recent_changes")
+
+    def get_tool_description(self):
+        return Tool(
+            name=self.name,
+            description="Get recently modified files in the vault.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "limit": {
+                        "type": "integer",
+                        "description": "Maximum number of files to return (default: 10)",
+                        "default": 10,
+                        "minimum": 1,
+                        "maximum": 100
+                    },
+                    "days": {
+                        "type": "integer",
+                        "description": "Only include files modified within this many days (optional)",
+                        "minimum": 1
+                    },
+                    "extensions": {
+                        "type": "array",
+                        "items": {
+                            "type": "string"
+                        },
+                        "description": "Only include files with these extensions (optional)"
+                    }
+                }
+            }
+        )
+
+    def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+        limit = args.get("limit", 10)
+        if not isinstance(limit, int) or limit < 1:
+            raise RuntimeError(f"Invalid limit: {limit}. Must be a positive integer")
+            
+        days = args.get("days")
+        if days is not None and (not isinstance(days, int) or days < 1):
+            raise RuntimeError(f"Invalid days: {days}. Must be a positive integer")
+            
+        extensions = args.get("extensions")
+        if extensions is not None and not isinstance(extensions, list):
+            raise RuntimeError(f"Invalid extensions: {extensions}. Must be an array of strings")
+
+        api = obsidian.Obsidian(api_key=api_key)
+        results = api.get_recent_changes(limit, days, extensions)
+
+        return [
+            TextContent(
+                type="text",
+                text=json.dumps(results, indent=2)
+            )
+        ]