""" Orchestrator tool registry. Each tool has two parts: 1. A Gemini FunctionDeclaration — tells the model what the tool does and what args it takes 2. A Python async callable — the actual implementation To add a new tool: 1. Implement it in a tools/.py module 2. Import it here and add (declaration, callable) to _REGISTRY 3. Add a FunctionDeclaration below and include it in TOOL_DECLARATIONS IMPORTANT: These tools are separate from the ae_* MCP tools used by the fleet agents. Do not modify the ae_* MCP server to support orchestrator needs. """ from google.genai import types from tools.web import search as _web_search from tools.ae_knowledge import journal_search as _ae_journal_search from tools.ae_knowledge import journal_list as _ae_journal_list from tools.ae_knowledge import journal_entry_read as _ae_journal_entry_read from tools.ae_knowledge import journal_entries_list as _ae_journal_entries_list from tools.ae_knowledge import journal_entry_create as _ae_journal_entry_create from tools.ae_knowledge import journal_entry_update as _ae_journal_entry_update from tools.ae_knowledge import journal_entry_disable as _ae_journal_entry_disable from tools.ae_knowledge import journal_entry_append as _ae_journal_entry_append from tools.ae_knowledge import journal_entry_prepend as _ae_journal_entry_prepend from tools.ae_tasks import task_list as _ae_task_list from tools.files import file_read as _file_read from tools.system import claude_allow_dir as _claude_allow_dir, shell_exec as _shell_exec from tools.tasks import task_list as _task_list, task_create as _task_create from tools.tasks import task_update as _task_update, task_complete as _task_complete from tools.cron import ( cron_list as _cron_list, cron_add as _cron_add, cron_remove as _cron_remove, cron_toggle as _cron_toggle, ) from tools.reminders import ( reminders_add as _reminders_add, reminders_list as _reminders_list, reminders_remove as _reminders_remove, reminders_clear as _reminders_clear, ) from tools.scratch import ( scratch_read as _scratch_read, scratch_write as _scratch_write, scratch_append as _scratch_append, scratch_clear as _scratch_clear, ) from tools.system import cortex_restart as _cortex_restart, cortex_logs as _cortex_logs from tools.web import http_fetch as _http_fetch from tools.files import file_list as _file_list, file_write as _file_write from tools.notify import nc_talk_send as _nc_talk_send, email_send as _email_send # --------------------------------------------------------------------------- # Gemini function declarations # --------------------------------------------------------------------------- _web_search_declaration = types.FunctionDeclaration( name="web_search", description=( "Search the web for current information. Use this when you need up-to-date " "facts, news, documentation, or anything not in your training data." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "query": types.Schema( type=types.Type.STRING, description="The search query string", ), "max_results": types.Schema( type=types.Type.INTEGER, description="Number of results to return (default 5, max 10)", ), }, required=["query"], ), ) _ae_journal_list_declaration = types.FunctionDeclaration( name="ae_journal_list", description=( "List all Aether Journals available for this account. " "Returns each journal's name and id_random. " "Call this first when you need to write a new entry or scope a search to a specific journal " "and don't already know the journal's id." ), parameters=types.Schema(type=types.Type.OBJECT, properties={}), ) _ae_journal_search_declaration = types.FunctionDeclaration( name="ae_journal_search", description=( "Search Aether Journal entries. All parameters are optional — combine freely. " "Use 'query' for fulltext keyword search (supports boolean: +required -excluded \"phrase\"). " "Use 'tags' to filter by tag substring. Use 'date_from'/'date_to' for date ranges (YYYY-MM-DD). " "Always search before creating a new entry to avoid duplicates." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "query": types.Schema( type=types.Type.STRING, description="Fulltext keyword search. Supports boolean mode: +required -excluded \"exact phrase\".", ), "journal_id": types.Schema( type=types.Type.STRING, description="Scope results to a specific journal by its id_random. Omit to search all journals.", ), "tags": types.Schema( type=types.Type.STRING, description="Filter by tag substring (e.g. 'networking' matches entries tagged 'networking' or 'home-networking').", ), "type_code": types.Schema( type=types.Type.STRING, description="Filter by exact type_code (e.g. 'note', 'meeting', 'log').", ), "topic_code": types.Schema( type=types.Type.STRING, description="Filter by exact topic_code.", ), "date_from": types.Schema( type=types.Type.STRING, description="Return entries created on or after this date (YYYY-MM-DD).", ), "date_to": types.Schema( type=types.Type.STRING, description="Return entries created on or before this date (YYYY-MM-DD).", ), "sort_by": types.Schema( type=types.Type.STRING, description="Sort field: 'updated' (default), 'created', 'name', or 'priority'.", ), "sort_order": types.Schema( type=types.Type.STRING, description="Sort direction: 'desc' (default, newest first) or 'asc'.", ), "status": types.Schema( type=types.Type.INTEGER, description="Filter by exact status code.", ), "priority": types.Schema( type=types.Type.INTEGER, description="Filter by exact priority (1=low, 5=high).", ), "max_results": types.Schema( type=types.Type.INTEGER, description="Number of results per page (default 10).", ), "page": types.Schema( type=types.Type.INTEGER, description="Page number for pagination (default 1).", ), }, required=[], ), ) _ae_journal_entry_read_declaration = types.FunctionDeclaration( name="ae_journal_entry_read", description=( "Fetch the full content of a single journal entry by its id_random. " "Use this when you need to read an entry before editing it, or when search results " "don't show enough content. Returns title, journal, tags, summary, and full content." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "entry_id": types.Schema( type=types.Type.STRING, description="The id_random of the journal entry to read.", ), "max_content_chars": types.Schema( type=types.Type.INTEGER, description="Maximum characters of content to return (default 4000). Increase for long entries.", ), }, required=["entry_id"], ), ) _ae_journal_entries_list_declaration = types.FunctionDeclaration( name="ae_journal_entries_list", description=( "List entries in a specific journal, newest first. " "Use this to browse what's in a journal when you don't have a search keyword, " "or to find entries by browsing rather than searching. " "Returns numbered entries with id, title, tags, summary, and date." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "journal_id": types.Schema( type=types.Type.STRING, description="The id_random of the journal to list entries from.", ), "max_results": types.Schema( type=types.Type.INTEGER, description="Number of entries to return (default 20, max 50).", ), "page": types.Schema( type=types.Type.INTEGER, description="Page number for pagination (default 1).", ), }, required=["journal_id"], ), ) _ae_journal_entry_create_declaration = types.FunctionDeclaration( name="ae_journal_entry_create", description=( "Create a new entry in an Aether Journal. " "Use this to save notes, summaries, or any content the user wants to store. " "Always call ae_journal_search first to check for existing entries on the same topic." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "journal_id": types.Schema( type=types.Type.STRING, description=( "The id_random of the target journal. " "Ask the user which journal to write to if not specified." ), ), "title": types.Schema( type=types.Type.STRING, description="Entry title", ), "content": types.Schema( type=types.Type.STRING, description="Full entry content (markdown supported)", ), "summary": types.Schema( type=types.Type.STRING, description="Optional short summary (1-2 sentences)", ), "tags": types.Schema( type=types.Type.STRING, description="Optional comma-separated tags (e.g. 'wireguard, networking, homelab')", ), }, required=["journal_id", "title", "content"], ), ) _ae_journal_entry_update_declaration = types.FunctionDeclaration( name="ae_journal_entry_update", description=( "Update fields on an existing journal entry. Only the fields you provide are changed — " "omitted fields are left as-is. Use ae_journal_search to find the entry_id first. " "To soft-delete, use ae_journal_entry_disable instead." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "entry_id": types.Schema(type=types.Type.STRING, description="Journal entry id_random"), "title": types.Schema(type=types.Type.STRING, description="New title"), "content": types.Schema(type=types.Type.STRING, description="Replacement content (full, markdown supported)"), "summary": types.Schema(type=types.Type.STRING, description="New summary"), "tags": types.Schema(type=types.Type.STRING, description="Replacement comma-separated tags"), "enable": types.Schema(type=types.Type.BOOLEAN, description="Set false to hide/disable the entry"), }, required=["entry_id"], ), ) _ae_journal_entry_disable_declaration = types.FunctionDeclaration( name="ae_journal_entry_disable", description=( "Soft-delete a journal entry by setting enable=false. " "The entry is hidden but not permanently removed. " "Use ae_journal_search to find the entry_id first." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "entry_id": types.Schema(type=types.Type.STRING, description="Journal entry id_random"), }, required=["entry_id"], ), ) _ae_journal_entry_append_declaration = types.FunctionDeclaration( name="ae_journal_entry_append", description=( "Append a new section to the bottom of a journal entry's content. " "Each section gets a UTC timestamp heading unless you provide one. " "Ideal for timestamped logs, running notes, or data logs." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "entry_id": types.Schema(type=types.Type.STRING, description="Journal entry id_random"), "content": types.Schema(type=types.Type.STRING, description="The text to append (markdown supported)"), "heading": types.Schema(type=types.Type.STRING, description="Optional section heading (defaults to current UTC timestamp)"), }, required=["entry_id", "content"], ), ) _ae_journal_entry_prepend_declaration = types.FunctionDeclaration( name="ae_journal_entry_prepend", description=( "Prepend a new section to the top of a journal entry's content. " "Each section gets a UTC timestamp heading unless you provide one. " "Useful for most-recent-first logs." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "entry_id": types.Schema(type=types.Type.STRING, description="Journal entry id_random"), "content": types.Schema(type=types.Type.STRING, description="The text to prepend (markdown supported)"), "heading": types.Schema(type=types.Type.STRING, description="Optional section heading (defaults to current UTC timestamp)"), }, required=["entry_id", "content"], ), ) _ae_task_list_declaration = types.FunctionDeclaration( name="ae_task_list", description=( "List tasks from the agents_sync Kanban board (todo and in-progress). " "Use this when asked about current work, pending tasks, or project status." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "include_done": types.Schema( type=types.Type.BOOLEAN, description="If true, also include completed tasks (default false)", ), }, ), ) _file_read_declaration = types.FunctionDeclaration( name="file_read", description=( "Read a local file and return its contents. " "Allowed directories: ~/agents_sync/, ~/OSIT_dev/, ~/DgrZone_Nextcloud/, ~/OSIT_Nextcloud/. " "Use this to read documentation, notes, CLAUDE.md files, or config references. " "If given a directory path, returns a directory listing instead." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "path": types.Schema( type=types.Type.STRING, description=( "Absolute or home-relative path to the file " "(e.g. ~/agents_sync/CLAUDE.md or /home/scott/agents_sync/tasks/01_todo/)" ), ), "max_lines": types.Schema( type=types.Type.INTEGER, description="Optional line limit (default 500)", ), }, required=["path"], ), ) # --------------------------------------------------------------------------- # Registry: maps tool name → async callable # --------------------------------------------------------------------------- _CALLABLES: dict[str, callable] = { "web_search": _web_search, "ae_journal_list": _ae_journal_list, "ae_journal_search": _ae_journal_search, "ae_journal_entry_read": _ae_journal_entry_read, "ae_journal_entries_list": _ae_journal_entries_list, "ae_journal_entry_create": _ae_journal_entry_create, "ae_journal_entry_update": _ae_journal_entry_update, "ae_journal_entry_disable": _ae_journal_entry_disable, "ae_journal_entry_append": _ae_journal_entry_append, "ae_journal_entry_prepend": _ae_journal_entry_prepend, "ae_task_list": _ae_task_list, "file_read": _file_read, "file_list": _file_list, "file_write": _file_write, "claude_allow_dir": _claude_allow_dir, "shell_exec": _shell_exec, "cortex_restart": _cortex_restart, "cortex_logs": _cortex_logs, "http_fetch": _http_fetch, "email_send": _email_send, "nc_talk_send": _nc_talk_send, "task_list": _task_list, "task_create": _task_create, "task_update": _task_update, "task_complete": _task_complete, "cron_list": _cron_list, "cron_add": _cron_add, "cron_remove": _cron_remove, "cron_toggle": _cron_toggle, "reminders_add": _reminders_add, "reminders_list": _reminders_list, "reminders_remove": _reminders_remove, "reminders_clear": _reminders_clear, "scratch_read": _scratch_read, "scratch_write": _scratch_write, "scratch_append": _scratch_append, "scratch_clear": _scratch_clear, } _claude_allow_dir_declaration = types.FunctionDeclaration( name="claude_allow_dir", description=( "Add a directory to Claude Code's auto-allow list so Claude can read or write " "files there without prompting. Edits ~/.claude/settings.json on the local machine. " "Use this when Claude is silently hanging or being blocked from accessing a directory. " "Changes take effect in the next Claude Code session." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "path": types.Schema( type=types.Type.STRING, description=( "Absolute or home-relative path to the directory " "(e.g. ~/OSIT_dev/aether_api_fastapi or /home/scott/agents_sync)" ), ), "mode": types.Schema( type=types.Type.STRING, description="Permission mode: 'r' (read-only), 'w' (write-only), or 'rw' (both). Default: rw", ), }, required=["path"], ), ) _shell_exec_declaration = types.FunctionDeclaration( name="shell_exec", description=( "Execute a shell command on the Cortex host machine and return its output. " "Use for system diagnostics: disk usage (df -h), process status (ps aux), " "directory listings (ls), memory (free -h), uptime, network info, log tails, etc. " "Commands run as the Cortex service user. Timeout enforced (default 30s, max 120s). " "Avoid destructive commands — prefer read-only system queries." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "command": types.Schema( type=types.Type.STRING, description="Shell command to run (e.g. 'df -h', 'ls ~/agents_sync/', 'journalctl --user -u cortex -n 50')", ), "working_dir": types.Schema( type=types.Type.STRING, description="Optional working directory (e.g. '~/agents_sync/projects'). Defaults to home directory.", ), "timeout": types.Schema( type=types.Type.INTEGER, description="Timeout in seconds (default 30, max 120)", ), }, required=["command"], ), ) _task_list_declaration = types.FunctionDeclaration( name="task_list", description=( "List personal tasks from Inara's task list. " "Use this to check what's on the list, review pending work, or find a task ID. " "Optionally filter by status: 'todo', 'in_progress', or 'done'." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "status": types.Schema( type=types.Type.STRING, description="Filter by status: 'todo', 'in_progress', or 'done'. Omit to list all.", ), }, ), ) _task_create_declaration = types.FunctionDeclaration( name="task_create", description=( "Add a new task to Inara's personal task list. " "Use this when the user asks to remember something, add a to-do, or track a follow-up." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "title": types.Schema( type=types.Type.STRING, description="Short task title", ), "description": types.Schema( type=types.Type.STRING, description="Optional longer description or context", ), "priority": types.Schema( type=types.Type.STRING, description="Priority: 'low', 'normal', or 'high'. Default: normal.", ), }, required=["title"], ), ) _task_update_declaration = types.FunctionDeclaration( name="task_update", description=( "Update an existing task. Use task_list first to get the task ID. " "Can update status, title, description, or priority. " "To just mark complete, use task_complete instead." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "task_id": types.Schema( type=types.Type.STRING, description="Task ID (e.g. t_abc123) — get from task_list", ), "status": types.Schema( type=types.Type.STRING, description="New status: 'todo', 'in_progress', or 'done'", ), "title": types.Schema( type=types.Type.STRING, description="Updated title", ), "description": types.Schema( type=types.Type.STRING, description="Updated description", ), "priority": types.Schema( type=types.Type.STRING, description="Updated priority: 'low', 'normal', or 'high'", ), }, required=["task_id"], ), ) _task_complete_declaration = types.FunctionDeclaration( name="task_complete", description=( "Mark a task as done. Use task_list first to get the task ID. " "Shorthand for task_update with status='done'." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "task_id": types.Schema( type=types.Type.STRING, description="Task ID (e.g. t_abc123) — get from task_list", ), }, required=["task_id"], ), ) _cron_list_declaration = types.FunctionDeclaration( name="cron_list", description=( "List all scheduled cron jobs — their ID, label, schedule, type, and last run time. " "Use this to see what's scheduled before adding or removing jobs." ), parameters=types.Schema(type=types.Type.OBJECT, properties={}), ) _cron_add_declaration = types.FunctionDeclaration( name="cron_add", description=( "Create a new scheduled cron job and register it immediately (no restart needed). " "Two types: 'remind' writes to the pending reminders queue (Inara sees it automatically " "in context next session); 'note' appends to the scratchpad. " "Schedule formats: 'hourly' | 'daily' | 'daily:HH:MM' | 'weekly:DOW' | 'weekly:DOW:HH:MM'. " "Example: schedule='daily:09:00', type='remind', payload='Check in with Scott.'" ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "label": types.Schema( type=types.Type.STRING, description="Short human-readable name for this job (e.g. 'Morning check-in')", ), "schedule": types.Schema( type=types.Type.STRING, description=( "When to run. Formats: hourly | daily | daily:HH:MM | " "weekly:DOW | weekly:DOW:HH:MM (e.g. 'weekly:mon:09:00')" ), ), "job_type": types.Schema( type=types.Type.STRING, description="'remind' (→ REMINDERS.md, auto-surfaced in context) or 'note' (→ SCRATCH.md)", ), "payload": types.Schema( type=types.Type.STRING, description="The text to write when the job fires", ), }, required=["label", "schedule", "job_type", "payload"], ), ) _cron_remove_declaration = types.FunctionDeclaration( name="cron_remove", description=( "Permanently delete a scheduled cron job. Use cron_list first to get the ID. " "To temporarily disable without deleting, use cron_toggle instead." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "cron_id": types.Schema( type=types.Type.STRING, description="Job ID (e.g. c_abc123) — get from cron_list", ), }, required=["cron_id"], ), ) _cron_toggle_declaration = types.FunctionDeclaration( name="cron_toggle", description=( "Pause a running cron job, or resume a paused one. " "The job stays in the list and can be re-enabled later. " "Use cron_list to see current enabled/paused state." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "cron_id": types.Schema( type=types.Type.STRING, description="Job ID (e.g. c_abc123) — get from cron_list", ), }, required=["cron_id"], ), ) _reminders_add_declaration = types.FunctionDeclaration( name="reminders_add", description=( "Add a new reminder to REMINDERS.md. Reminders are automatically surfaced " "in your context at the start of each session (Tier 2+). " "Use this when the user asks you to remember something, follow up on something, " "or surface a note at the next session." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "text": types.Schema( type=types.Type.STRING, description="The reminder text to add", ), "label": types.Schema( type=types.Type.STRING, description="Optional heading for this reminder (e.g. 'Follow up on NC Talk'). Defaults to current timestamp.", ), }, required=["text"], ), ) _reminders_list_declaration = types.FunctionDeclaration( name="reminders_list", description=( "Read all current pending reminders from REMINDERS.md. " "Use this to check what reminders are queued before adding duplicates, " "or to show the user what's pending." ), parameters=types.Schema(type=types.Type.OBJECT, properties={}), ) _reminders_remove_declaration = types.FunctionDeclaration( name="reminders_remove", description=( "Remove a single reminder by its number. " "Call reminders_list first to get the numbered list, then pass the number of the reminder to remove." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "index": types.Schema( type=types.Type.INTEGER, description="The number of the reminder to remove (1 = first item in reminders_list output).", ), }, required=["index"], ), ) _reminders_clear_declaration = types.FunctionDeclaration( name="reminders_clear", description=( "Erase all pending reminders from REMINDERS.md. " "Use this after you have acknowledged and acted on the reminders shown in your context." ), parameters=types.Schema(type=types.Type.OBJECT, properties={}), ) _scratch_read_declaration = types.FunctionDeclaration( name="scratch_read", description=( "Read the full contents of the scratchpad. " "Use this to recall working notes, mid-task context, or anything previously jotted down. " "The scratchpad is transient — nothing here is distilled or archived." ), parameters=types.Schema(type=types.Type.OBJECT, properties={}), ) _scratch_write_declaration = types.FunctionDeclaration( name="scratch_write", description=( "Replace the entire scratchpad with new content. " "Use this to set a clean working note, replacing whatever was there before. " "For adding without replacing, use scratch_append instead." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "content": types.Schema( type=types.Type.STRING, description="The new scratchpad content (markdown supported)", ), }, required=["content"], ), ) _scratch_append_declaration = types.FunctionDeclaration( name="scratch_append", description=( "Add a new section to the bottom of the scratchpad without replacing existing content. " "Each section gets a timestamp heading unless you supply one." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "content": types.Schema( type=types.Type.STRING, description="The content to append (markdown supported)", ), "heading": types.Schema( type=types.Type.STRING, description="Optional section heading. Defaults to current UTC timestamp.", ), }, required=["content"], ), ) _scratch_clear_declaration = types.FunctionDeclaration( name="scratch_clear", description="Erase everything in the scratchpad. Use when the working notes are no longer needed.", parameters=types.Schema(type=types.Type.OBJECT, properties={}), ) _cortex_restart_declaration = types.FunctionDeclaration( name="cortex_restart", description=( "Restart the Cortex service via systemd. Schedules a restart 5 seconds from now. " "The current connection will drop — inform the user to refresh the page. " "Use after config changes, memory edits, or when the service needs a fresh start. " "ADMIN ONLY." ), parameters=types.Schema(type=types.Type.OBJECT, properties={}), ) _cortex_logs_declaration = types.FunctionDeclaration( name="cortex_logs", description=( "Fetch recent lines from the Cortex systemd service journal. " "Use for debugging errors, checking startup status, or reviewing recent activity. " "ADMIN ONLY." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "lines": types.Schema( type=types.Type.INTEGER, description="Number of log lines to return (default 50, max 200)", ), }, ), ) _http_fetch_declaration = types.FunctionDeclaration( name="http_fetch", description=( "Fetch a specific URL and return the response. Unlike web_search, this hits " "a direct URL — useful for health checks, JSON API endpoints, webhook testing, " "or reading a specific page when you already know the URL. " "Response body is capped at 8 KB." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "url": types.Schema(type=types.Type.STRING, description="Full URL to fetch"), "method": types.Schema( type=types.Type.STRING, description="HTTP method: GET (default), POST, HEAD", ), "body": types.Schema( type=types.Type.STRING, description="Optional request body (for POST requests)", ), "timeout": types.Schema( type=types.Type.INTEGER, description="Request timeout in seconds (default 15, max 60)", ), }, required=["url"], ), ) _file_list_declaration = types.FunctionDeclaration( name="file_list", description=( "List the files and subdirectories in a directory. " "Allowed paths: ~/agents_sync/, ~/OSIT_dev/, ~/DgrZone_Nextcloud/, ~/OSIT_Nextcloud/. " "ADMIN ONLY." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "path": types.Schema( type=types.Type.STRING, description="Absolute or home-relative path to the directory", ), }, required=["path"], ), ) _file_write_declaration = types.FunctionDeclaration( name="file_write", description=( "Write or append content to a file. " "Write-allowed paths: ~/agents_sync/ and the Cortex home/ directory. " "Creates parent directories if needed. " "ADMIN ONLY. Requires user confirmation before executing." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "path": types.Schema( type=types.Type.STRING, description="Absolute or home-relative path to write to", ), "content": types.Schema( type=types.Type.STRING, description="Content to write", ), "mode": types.Schema( type=types.Type.STRING, description="'overwrite' (default, replaces file) or 'append' (adds to end)", ), }, required=["path", "content"], ), ) _email_send_declaration = types.FunctionDeclaration( name="email_send", description=( "Send an email from the server's configured SMTP account. Use for delivering " "summaries, reports, reminders, or any content the user wants emailed. " "body is plain text; newlines are preserved." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "to": types.Schema(type=types.Type.STRING, description="Recipient email address"), "subject": types.Schema(type=types.Type.STRING, description="Email subject line"), "body": types.Schema(type=types.Type.STRING, description="Plain-text email body"), }, required=["to", "subject", "body"], ), ) _nc_talk_send_declaration = types.FunctionDeclaration( name="nc_talk_send", description=( "Send a proactive message to the user via their configured notification channel " "(Nextcloud Talk by default). Use this to notify the user of completed background " "tasks, important events, or anything they should know between sessions. " "Requires notification_channel and notification_room set in channels.json." ), parameters=types.Schema( type=types.Type.OBJECT, properties={ "message": types.Schema( type=types.Type.STRING, description="The message to send to the user", ), }, required=["message"], ), ) # --------------------------------------------------------------------------- # Role-based access control # --------------------------------------------------------------------------- # Minimum role required to use each tool. Unlisted tools default to "user". TOOL_ROLES: dict[str, str] = { # Admin-only — system-level or broad filesystem access "shell_exec": "admin", "claude_allow_dir": "admin", "cortex_restart": "admin", "cortex_logs": "admin", "file_read": "admin", "file_list": "admin", "file_write": "admin", "ae_task_list": "admin", # reads agents_sync kanban "email_send": "admin", # sends from server SMTP account "nc_talk_send": "admin", } # Tools that require explicit user confirmation before executing. # The orchestrator injects a CONFIRMATION_REQUIRED result instead of calling # the tool, prompting Claude to ask the user to confirm in a follow-up message. CONFIRM_REQUIRED: set[str] = { "cortex_restart", "file_write", "shell_exec", "cron_remove", "reminders_clear", } _ROLE_RANK: dict[str, int] = {"user": 0, "admin": 1} def _role_allowed(tool_name: str, role: str) -> bool: required = TOOL_ROLES.get(tool_name, "user") return _ROLE_RANK.get(role, 0) >= _ROLE_RANK.get(required, 0) # Flat list of all declarations — single source of truth for both Gemini and OpenAI formats. _ALL_DECLARATIONS: list[types.FunctionDeclaration] = [ _web_search_declaration, _ae_journal_list_declaration, _ae_journal_search_declaration, _ae_journal_entry_read_declaration, _ae_journal_entries_list_declaration, _ae_journal_entry_create_declaration, _ae_journal_entry_update_declaration, _ae_journal_entry_disable_declaration, _ae_journal_entry_append_declaration, _ae_journal_entry_prepend_declaration, _ae_task_list_declaration, _file_read_declaration, _file_list_declaration, _file_write_declaration, _claude_allow_dir_declaration, _shell_exec_declaration, _cortex_restart_declaration, _cortex_logs_declaration, _http_fetch_declaration, _email_send_declaration, _nc_talk_send_declaration, _task_list_declaration, _task_create_declaration, _task_update_declaration, _task_complete_declaration, _cron_list_declaration, _cron_add_declaration, _cron_remove_declaration, _cron_toggle_declaration, _reminders_add_declaration, _reminders_list_declaration, _reminders_remove_declaration, _reminders_clear_declaration, _scratch_read_declaration, _scratch_write_declaration, _scratch_append_declaration, _scratch_clear_declaration, ] # Full Gemini Tool object (all tools — use get_tools_for_role() in production) TOOL_DECLARATIONS = [types.Tool(function_declarations=_ALL_DECLARATIONS)] async def call_tool(name: str, args: dict, callables: dict | None = None) -> str: """Dispatch a tool call by name. Returns result as a string. Pass `callables` (from get_tools_for_role) to enforce role restrictions. Falls back to the full _CALLABLES dict if omitted. """ dispatch = callables if callables is not None else _CALLABLES fn = dispatch.get(name) if fn is None: return f"Tool not available or access denied: {name}" return await fn(**args) # --------------------------------------------------------------------------- # OpenAI JSON Schema format — auto-derived from the Gemini declarations above # so there is a single source of truth for tool definitions. # --------------------------------------------------------------------------- _GEMINI_TYPE_TO_JSON = { "OBJECT": "object", "STRING": "string", "INTEGER": "integer", "NUMBER": "number", "BOOLEAN": "boolean", "ARRAY": "array", } def _schema_to_json(schema) -> dict: """Recursively convert a Gemini types.Schema to a JSON Schema dict.""" type_name = getattr(getattr(schema, "type", None), "name", "STRING") result: dict = {"type": _GEMINI_TYPE_TO_JSON.get(type_name, "string")} if getattr(schema, "description", None): result["description"] = schema.description props = getattr(schema, "properties", None) or {} if result["type"] == "object": result["properties"] = {k: _schema_to_json(v) for k, v in props.items()} req = getattr(schema, "required", None) if req: result["required"] = list(req) return result def _build_openai_tools() -> list[dict]: """Convert _ALL_DECLARATIONS (Gemini format) to OpenAI tool schemas.""" out = [] for decl in _ALL_DECLARATIONS: params = ( _schema_to_json(decl.parameters) if decl.parameters else {"type": "object", "properties": {}} ) out.append({ "type": "function", "function": { "name": decl.name, "description": decl.description or "", "parameters": params, }, }) return out # OpenAI-format tool list — all tools (use get_openai_tools_for_role() in production) OPENAI_TOOL_SCHEMAS: list[dict] = _build_openai_tools() # --------------------------------------------------------------------------- # Role-filtered tool access # --------------------------------------------------------------------------- def get_tools_for_role(role: str) -> tuple[list, dict]: """Return (gemini_tool_declarations, callables_dict) filtered to tools the role can use. Usage in orchestrator: tool_declarations, tool_callables = get_tools_for_role(user_role) """ allowed = {name for name in _CALLABLES if _role_allowed(name, role)} decls = [d for d in _ALL_DECLARATIONS if d.name in allowed] callables = {k: v for k, v in _CALLABLES.items() if k in allowed} return [types.Tool(function_declarations=decls)], callables def get_openai_tools_for_role(role: str) -> list[dict]: """Return OpenAI tool schemas filtered to tools the role can use.""" allowed = {name for name in _CALLABLES if _role_allowed(name, role)} return [t for t in OPENAI_TOOL_SCHEMAS if t["function"]["name"] in allowed]