from __future__ import annotations import os from typing import Annotated, Optional import gradio as gr from app import _log_call_end, _log_call_start, _truncate_for_log from ._docstrings import autodoc from ._core import obsidian_sandbox, OBSIDIAN_ROOT TOOL_SUMMARY = ( "Browse and search the Obsidian vault in read-only mode. " "Actions: list, read, info, search, help. " "All paths resolve within the vault root. Start paths with '/' (e.g., /Notes)." ) HELP_TEXT = ( "Obsidian Vault — actions and usage\n\n" "Root: Tools/Obsidian (override with OBSIDIAN_VAULT_ROOT). " "Start paths with '/' to reference the vault root (e.g., /Projects/note.md). " "Absolute paths are disabled unless UNSAFE_ALLOW_ABS_PATHS=1.\n\n" "Actions and fields:\n" "- list: path='/' (default), recursive=false, show_hidden=false, max_entries=20\n" "- read: path (e.g., /Projects/note.md), offset=0, max_chars=4000 (shows next_cursor when truncated)\n" "- info: path\n" "- search: path (note or folder), query text in the Search field, recursive=false, show_hidden=false, max_entries=20, case_sensitive=false, offset=0\n" "- help: show this guide\n\n" "Errors are returned as JSON with fields: {status:'error', code, message, path?, hint?, data?}.\n\n" "Examples:\n" "- list current: action=list, path='/'\n" "- read note: action=read, path='/Projects/note.md', max_chars=500\n" "- show metadata: action=info, path='/Inbox'\n" "- search notes: action=search, path='/Projects', query='deadline', recursive=true, max_entries=100\n" "- case-sensitive search: action=search, query='TODO', case_sensitive=true\n" "- page search results: action=search, query='TODO', offset=20\n" ) # Use the pre-configured Obsidian sandbox _sandbox = obsidian_sandbox ROOT_DIR = OBSIDIAN_ROOT # Convenience wrappers def _resolve_path(path: str) -> tuple[str, str]: return _sandbox.resolve_path(path) def _display_path(abs_path: str) -> str: return _sandbox.display_path(abs_path) def _err(code: str, message: str, *, path: str | None = None, hint: str | None = None, data: dict | None = None) -> str: return _sandbox.err(code, message, path=path, hint=hint, data=data) @autodoc(summary=TOOL_SUMMARY) def Obsidian_Vault( action: Annotated[str, "Operation to perform: 'list', 'read', 'info', 'search', 'help'."], path: Annotated[str, "Target path, relative to the vault root." ] = "/", query: Annotated[Optional[str], "Text to search for when action=search."] = None, recursive: Annotated[bool, "Recurse into subfolders when listing/searching."] = False, show_hidden: Annotated[bool, "Include hidden files when listing/searching."] = False, max_entries: Annotated[int, "Max entries to list or matches to return (for list/search)."] = 20, offset: Annotated[int, "Start offset when reading files."] = 0, max_chars: Annotated[int, "Max characters to return when reading (0 = full file)."] = 4000, case_sensitive: Annotated[bool, "Match case when searching text."] = False, ) -> str: _log_call_start( "Obsidian_Vault", action=action, path=path, query=query, recursive=recursive, show_hidden=show_hidden, max_entries=max_entries, offset=offset, max_chars=max_chars, case_sensitive=case_sensitive, ) action = (action or "").strip().lower() if action not in {"list", "read", "info", "search", "help"}: result = _err( "invalid_action", "Invalid action.", hint="Choose from: list, read, info, search, help.", ) _log_call_end("Obsidian_Vault", _truncate_for_log(result)) return result if action == "help": result = HELP_TEXT _log_call_end("Obsidian_Vault", _truncate_for_log(result)) return result abs_path, err = _resolve_path(path) if err: _log_call_end("Obsidian_Vault", _truncate_for_log(err)) return err try: if action == "list": if not os.path.exists(abs_path): result = _err("path_not_found", f"Path not found: {_display_path(abs_path)}", path=_display_path(abs_path)) else: result = _sandbox.list_dir(abs_path, show_hidden=show_hidden, recursive=recursive, max_entries=max_entries) elif action == "read": result = _sandbox.read_file(abs_path, offset=offset, max_chars=max_chars) elif action == "search": query_text = query or "" if query_text.strip() == "": result = _err( "missing_search_query", "Search query is required for the search action.", hint="Provide text in the Search field to look for.", ) else: result = _sandbox.search_text( abs_path, query_text, recursive=recursive, show_hidden=show_hidden, max_results=max_entries, case_sensitive=case_sensitive, start_index=offset, ) else: # info result = _sandbox.info(abs_path) except Exception as exc: result = _err("exception", "Unhandled error during operation.", data={"error": _sandbox.safe_err(exc)}) _log_call_end("Obsidian_Vault", _truncate_for_log(result)) return result def build_interface() -> gr.Interface: return gr.Interface( fn=Obsidian_Vault, inputs=[ gr.Radio( label="Action", choices=["list", "read", "info", "search", "help"], value="help", info="Operation to perform", ), gr.Textbox(label="Path", placeholder="/ or /Notes/todo.md", max_lines=1, value="/", info="Target path (relative to vault root)"), gr.Textbox(label="Search text", lines=3, placeholder="Text to search for...", info="Text to search for (Search only)"), gr.Checkbox(label="Recursive", value=False, info="Recurse into subfolders (List/Search)"), gr.Checkbox(label="Show hidden", value=False, info="Include hidden files (List/Search)"), gr.Slider(minimum=10, maximum=5000, step=10, value=20, label="Max entries / matches", info="Max entries to list or matches to return (List/Search)"), gr.Slider(minimum=0, maximum=1_000_000, step=100, value=0, label="Offset", info="Start offset (Read/Search)"), gr.Slider(minimum=0, maximum=100_000, step=500, value=4000, label="Max chars", info="Max characters to return (Read, 0=all)"), gr.Checkbox(label="Case sensitive search", value=False, info="Match case (Search)"), ], outputs=gr.Textbox(label="Result", lines=20), title="Obsidian Vault", description=( "
Explore and search notes in the vault without modifying them." "
" ), api_description=TOOL_SUMMARY, flagging_mode="never", submit_btn="Run", ) __all__ = ["Obsidian_Vault", "build_interface"]