""" title: Obsidian REST API Access author: TSTXD777 description: Access and manage your Obsidian notes through the Local REST API required_open_webui_version: 0.4.0 requirements: requests, urllib3 version: 1.0.0 licence: MIT """ import requests import json import urllib.parse from datetime import datetime from typing import Optional, List, Dict, Any, Union from pydantic import BaseModel, Field import urllib3 # Disable SSL warnings for self-signed certificates urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) class Tools: def __init__(self): """Initialize the Obsidian Notes Tool.""" self.valves = self.Valves() self.citation = False class Valves(BaseModel): api_key: str = Field( default="", description="Your Obsidian Local REST API key (found in Obsidian settings under 'Local REST API')", ) base_url: str = Field( default="https://127.0.0.1:27124", description="Base URL for your Obsidian Local REST API (default: https://127.0.0.1:27124)", ) verify_ssl: bool = Field( default=False, description="Verify SSL certificates (set to False for self-signed certificates)", ) def _get_headers(self) -> Dict[str, str]: """Get headers for API requests.""" return { "Authorization": f"Bearer {self.valves.api_key}", "Content-Type": "application/json", } def _make_request(self, method: str, endpoint: str, **kwargs) -> requests.Response: """Make a request to the Obsidian API.""" url = f"{self.valves.base_url.rstrip('/')}{endpoint}" headers = self._get_headers() if "headers" in kwargs: headers.update(kwargs["headers"]) kwargs["headers"] = headers kwargs["verify"] = self.valves.verify_ssl return requests.request(method, url, **kwargs) def get_server_info(self) -> str: """ Get basic server information and authentication status. This is useful to test if your API connection is working properly. """ try: response = self._make_request("GET", "/") if response.status_code == 200: data = response.json() return ( f"✅ Connected to Obsidian Local REST API\n" f"Service: {data.get('service', 'Unknown')}\n" f"Authenticated: {data.get('authenticated', False)}\n" f"Plugin Version: {data.get('versions', {}).get('self', 'Unknown')}\n" f"Obsidian API Version: {data.get('versions', {}).get('obsidian', 'Unknown')}" ) else: return f"❌ Failed to connect: {response.status_code} - {response.text}" except Exception as e: return f"❌ Connection error: {str(e)}" def get_active_file(self, json_format: bool = False) -> str: """ Get the content of the currently active file in Obsidian. :param json_format: If True, returns JSON with metadata, tags, and frontmatter. If False, returns raw markdown. """ try: headers = {} if json_format: headers["Accept"] = "application/vnd.olrapi.note+json" else: headers["Accept"] = "text/markdown" response = self._make_request("GET", "/active/", headers=headers) if response.status_code == 200: if json_format: data = response.json() return ( f"📄 **Active File:** {data.get('path', 'Unknown')}\n\n" f"**Tags:** {', '.join(data.get('tags', []))}\n" f"**Created:** {datetime.fromtimestamp(data.get('stat', {}).get('ctime', 0) / 1000)}\n" f"**Modified:** {datetime.fromtimestamp(data.get('stat', {}).get('mtime', 0) / 1000)}\n\n" f"**Content:**\n```markdown\n{data.get('content', '')}\n```" ) else: return f"📄 **Active File Content:**\n```markdown\n{response.text}\n```" elif response.status_code == 404: return "❌ No active file found in Obsidian" else: return f"❌ Error: {response.status_code} - {response.text}" except Exception as e: return f"❌ Error getting active file: {str(e)}" def create_or_update_file(self, file_path: str, content: str) -> str: """ Create a new file or completely replace the content of an existing file. :param file_path: Path to the file relative to vault root (e.g., "folder/note.md") :param content: The complete content for the file in markdown format """ try: response = self._make_request( "PUT", f"/vault/{urllib.parse.quote(file_path, safe='/')}", data=content, headers={"Content-Type": "text/markdown"}, ) if response.status_code == 204: return f"✅ Successfully created/updated file: {file_path}" else: return f"❌ Error creating/updating file: {response.status_code} - {response.text}" except Exception as e: return f"❌ Error: {str(e)}" def append_to_file(self, file_path: str, content: str) -> str: """ Append content to the end of an existing file. Creates the file if it doesn't exist. :param file_path: Path to the file relative to vault root (e.g., "folder/note.md") :param content: Content to append to the file """ try: response = self._make_request( "POST", f"/vault/{urllib.parse.quote(file_path, safe='/')}", data=content, headers={"Content-Type": "text/markdown"}, ) if response.status_code == 204: return f"✅ Successfully appended content to: {file_path}" else: return f"❌ Error appending to file: {response.status_code} - {response.text}" except Exception as e: return f"❌ Error: {str(e)}" def get_file_content(self, file_path: str, json_format: bool = False) -> str: """ Get the content of a specific file in your vault. :param file_path: Path to the file relative to vault root (e.g., "folder/note.md") :param json_format: If True, returns JSON with metadata, tags, and frontmatter. If False, returns raw markdown. """ try: headers = {} if json_format: headers["Accept"] = "application/vnd.olrapi.note+json" else: headers["Accept"] = "text/markdown" response = self._make_request( "GET", f"/vault/{urllib.parse.quote(file_path, safe='/')}", headers=headers, ) if response.status_code == 200: if json_format: data = response.json() return ( f"📄 **File:** {data.get('path', 'Unknown')}\n\n" f"**Tags:** {', '.join(data.get('tags', []))}\n" f"**Created:** {datetime.fromtimestamp(data.get('stat', {}).get('ctime', 0) / 1000)}\n" f"**Modified:** {datetime.fromtimestamp(data.get('stat', {}).get('mtime', 0) / 1000)}\n\n" f"**Content:**\n```markdown\n{data.get('content', '')}\n```" ) else: return f"📄 **File Content ({file_path}):**\n```markdown\n{response.text}\n```" elif response.status_code == 404: return f"❌ File not found: {file_path}" else: return f"❌ Error: {response.status_code} - {response.text}" except Exception as e: return f"❌ Error getting file: {str(e)}" def delete_file(self, file_path: str) -> str: """ Delete a specific file from your vault. :param file_path: Path to the file relative to vault root (e.g., "folder/note.md") """ try: response = self._make_request( "DELETE", f"/vault/{urllib.parse.quote(file_path, safe='/')}" ) if response.status_code == 204: return f"✅ Successfully deleted file: {file_path}" elif response.status_code == 404: return f"❌ File not found: {file_path}" else: return ( f"❌ Error deleting file: {response.status_code} - {response.text}" ) except Exception as e: return f"❌ Error: {str(e)}" def list_vault_files(self, directory_path: str = "") -> str: """ List files and directories in your vault. :param directory_path: Path to directory relative to vault root (empty string for root directory) """ try: if directory_path: endpoint = f"/vault/{urllib.parse.quote(directory_path, safe='/')}/" else: endpoint = "/vault/" response = self._make_request("GET", endpoint) if response.status_code == 200: data = response.json() files = data.get("files", []) if not files: return f"📁 Directory '{directory_path or 'root'}' is empty" result = f"📁 **Directory:** {directory_path or 'root'}\n\n" directories = [f for f in files if f.endswith("/")] markdown_files = [f for f in files if f.endswith(".md")] other_files = [ f for f in files if not f.endswith("/") and not f.endswith(".md") ] if directories: result += "**📁 Directories:**\n" for d in sorted(directories): result += f" • {d}\n" result += "\n" if markdown_files: result += "**📄 Markdown Files:**\n" for f in sorted(markdown_files): result += f" • {f}\n" result += "\n" if other_files: result += "**📎 Other Files:**\n" for f in sorted(other_files): result += f" • {f}\n" return result elif response.status_code == 404: return f"❌ Directory not found: {directory_path}" else: return f"❌ Error: {response.status_code} - {response.text}" except Exception as e: return f"❌ Error: {str(e)}" def patch_file_content( self, file_path: str, operation: str, target_type: str, target: str, content: str, ) -> str: """ Insert content into a file at a specific location (heading, block reference, or frontmatter). :param file_path: Path to the file relative to vault root :param operation: Operation to perform ('append', 'prepend', or 'replace') :param target_type: Type of target ('heading', 'block', or 'frontmatter') :param target: The target identifier (heading name, block ID, or frontmatter field) :param content: Content to insert """ try: headers = { "Operation": operation, "Target-Type": target_type, "Target": target, "Content-Type": "text/markdown", } response = self._make_request( "PATCH", f"/vault/{urllib.parse.quote(file_path, safe='/')}", data=content, headers=headers, ) if response.status_code == 200: return f"✅ Successfully {operation}ed content to {target_type} '{target}' in {file_path}" elif response.status_code == 404: return f"❌ File or target not found: {file_path} -> {target}" else: return f"❌ Error: {response.status_code} - {response.text}" except Exception as e: return f"❌ Error: {str(e)}" def search_vault(self, query: str, context_length: int = 100) -> str: """ Search for text across all files in your vault. :param query: Text to search for :param context_length: How much context to return around matches (default: 100 characters) """ try: params = {"query": query, "contextLength": context_length} response = self._make_request("POST", "/search/simple/", params=params) if response.status_code == 200: results = response.json() if not results: return f"🔍 No results found for: '{query}'" output = f"🔍 **Search Results for:** '{query}'\n\n" for result in results[:10]: # Limit to top 10 results filename = result.get("filename", "Unknown") score = result.get("score", 0) matches = result.get("matches", []) output += f"📄 **{filename}** (Score: {score:.2f})\n" for match in matches[:3]: # Show max 3 matches per file context = match.get("context", "") output += f" 💬 ...{context}...\n" output += "\n" return output else: return f"❌ Search error: {response.status_code} - {response.text}" except Exception as e: return f"❌ Error: {str(e)}" def get_daily_note(self, json_format: bool = False) -> str: """ Get today's daily note. Creates it if it doesn't exist. :param json_format: If True, returns JSON with metadata. If False, returns raw markdown. """ try: headers = {} if json_format: headers["Accept"] = "application/vnd.olrapi.note+json" else: headers["Accept"] = "text/markdown" response = self._make_request("GET", "/periodic/daily/", headers=headers) if response.status_code == 200: if json_format: data = response.json() return ( f"📅 **Daily Note:** {data.get('path', 'Unknown')}\n\n" f"**Content:**\n```markdown\n{data.get('content', '')}\n```" ) else: return ( f"📅 **Daily Note Content:**\n```markdown\n{response.text}\n```" ) elif response.status_code == 404: return "❌ Daily note not found (you may need to configure daily notes in Obsidian)" else: return f"❌ Error: {response.status_code} - {response.text}" except Exception as e: return f"❌ Error: {str(e)}" def append_to_daily_note(self, content: str) -> str: """ Append content to today's daily note. Creates the note if it doesn't exist. :param content: Content to append to the daily note """ try: response = self._make_request( "POST", "/periodic/daily/", data=content, headers={"Content-Type": "text/markdown"}, ) if response.status_code == 204: return f"✅ Successfully appended content to today's daily note" else: return f"❌ Error appending to daily note: {response.status_code} - {response.text}" except Exception as e: return f"❌ Error: {str(e)}" def open_file_in_obsidian(self, file_path: str, new_leaf: bool = False) -> str: """ Open a specific file in the Obsidian user interface. :param file_path: Path to the file relative to vault root :param new_leaf: Whether to open in a new tab/pane (default: False) """ try: params = {} if new_leaf: params["newLeaf"] = "true" response = self._make_request( "POST", f"/open/{urllib.parse.quote(file_path, safe='/')}", params=params, ) if response.status_code == 200: return f"✅ Successfully opened '{file_path}' in Obsidian" else: return ( f"❌ Error opening file: {response.status_code} - {response.text}" ) except Exception as e: return f"❌ Error: {str(e)}"