From c6bc8202b84290e18ee742f7d5d5876dd4d37e8e Mon Sep 17 00:00:00 2001 From: Brian Caudill Date: Mon, 1 Jun 2026 12:50:22 -0400 Subject: [PATCH] Add Agent Bridge addon Agent Bridge embeds a control bridge in a running Gramps session so an AI agent can drive the live application: read and modify the tree, operate the UI, and create and load plugins on the fly. It ships an MCP (Model Context Protocol) server so any MCP-capable AI can drive Gramps through standard tools. The gramplet polls a watched control directory on the GTK main thread and executes submitted Python in a persistent namespace with dbstate, db, uistate, gui and gramps_lib bound; the MCP server is a thin stdio adapter over that directory. No network port is opened. As defense-in-depth, every request must carry a shared secret token. The gramplet generates it on first run at ~/.gramps_agent/token (owner-only) and refuses any request without it; the MCP server and the debug CLI read the same file automatically. GRAMPS_AGENT_TOKEN overrides the file. --- AgentBridge/AgentBridge.gpr.py | 52 +++++ AgentBridge/AgentBridge.py | 347 +++++++++++++++++++++++++++++++ AgentBridge/MANIFEST | 1 + AgentBridge/README.md | 119 +++++++++++ AgentBridge/agent_send.py | 161 ++++++++++++++ AgentBridge/gramps_mcp_server.py | 235 +++++++++++++++++++++ AgentBridge/po/template.pot | 62 ++++++ 7 files changed, 977 insertions(+) create mode 100644 AgentBridge/AgentBridge.gpr.py create mode 100644 AgentBridge/AgentBridge.py create mode 100644 AgentBridge/MANIFEST create mode 100644 AgentBridge/README.md create mode 100644 AgentBridge/agent_send.py create mode 100644 AgentBridge/gramps_mcp_server.py create mode 100644 AgentBridge/po/template.pot diff --git a/AgentBridge/AgentBridge.gpr.py b/AgentBridge/AgentBridge.gpr.py new file mode 100644 index 000000000..5bd095721 --- /dev/null +++ b/AgentBridge/AgentBridge.gpr.py @@ -0,0 +1,52 @@ +# +# Gramps - a GTK+/GNOME based genealogy program +# +# Copyright (C) 2026 Brian Caudill +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along +# with this program; if not, see . +# + +# ------------------------------------------------------------------------ +# +# Register the Agent Bridge gramplet +# +# ------------------------------------------------------------------------ +register( + GRAMPLET, + id="Agent Bridge", + name=_("Agent Bridge"), + description=_( + "Embeds a control bridge in Gramps so an AI agent can drive the live " + "application through a watched directory or an MCP server. Executes " + "submitted Python on the GTK main thread. For developers and power " + "users; runs arbitrary code at your privileges." + ), + version="0.0.1", + gramps_target_version="6.0", + status=STABLE, + audience=DEVELOPER, + fname="AgentBridge.py", + gramplet="AgentBridge", + gramplet_title=_("Agent Bridge"), + height=140, + expand=True, + detached_width=520, + detached_height=300, + navtypes=["Dashboard"], + authors=["Brian Caudill"], + authors_email=["brian.m.caudill@gmail.com"], + maintainers=["Brian Caudill"], + maintainers_email=["brian.m.caudill@gmail.com"], + help_url="Addon:AgentBridge", +) diff --git a/AgentBridge/AgentBridge.py b/AgentBridge/AgentBridge.py new file mode 100644 index 000000000..7eca0e78b --- /dev/null +++ b/AgentBridge/AgentBridge.py @@ -0,0 +1,347 @@ +# +# Gramps - a GTK+/GNOME based genealogy program +# +# Copyright (C) 2026 Brian Caudill +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along +# with this program; if not, see . +# + +""" +Agent Bridge gramplet. + +Embeds a control bridge inside the running Gramps process so an external agent +(typically an AI via the bundled MCP server) can drive the live application. +The agent communicates through a watched control directory: it drops request +files, the bridge executes them on the GTK main thread and writes response +files. + +SECURITY: this executes arbitrary Python inside Gramps with the privileges of +the user running it. It listens on no network port -- control is purely via +files under ``~/.gramps_agent`` -- so anyone able to write to that directory +(i.e. this user account) can run code in Gramps. Only enable it on a machine +you trust, and remove the gramplet when you are done. +""" + +# ------------------------------------------------------------------------- +# +# Standard Python modules +# +# ------------------------------------------------------------------------- +import os +import io +import hmac +import json +import time +import stat +import secrets +import logging +import traceback +import contextlib + +# ------------------------------------------------------------------------- +# +# GTK/Gnome modules +# +# ------------------------------------------------------------------------- +from gi.repository import GLib + +# ------------------------------------------------------------------------- +# +# Gramps modules +# +# ------------------------------------------------------------------------- +from gramps.gen.plug import Gramplet +from gramps.gen.const import GRAMPS_LOCALE as glocale + +try: + _trans = glocale.get_addon_translator(__file__) +except ValueError: + _trans = glocale.translation +_ = _trans.gettext + +LOG = logging.getLogger("AgentBridge") + +# ------------------------------------------------------------------------- +# +# Constants +# +# ------------------------------------------------------------------------- +CONTROL_DIR = os.path.join(os.path.expanduser("~"), ".gramps_agent") +REQ_DIR = os.path.join(CONTROL_DIR, "requests") +RESP_DIR = os.path.join(CONTROL_DIR, "responses") +TOKEN_FILE = os.path.join(CONTROL_DIR, "token") +POLL_MS = 300 +MAX_REPR = 20000 + + +# ------------------------------------------------------------------------- +# +# Helper functions +# +# ------------------------------------------------------------------------- +def _ensure_token(): + """ + Return the shared secret, creating it on first run. + + An explicit ``GRAMPS_AGENT_TOKEN`` environment variable wins; otherwise the + token is read from (or generated into) ``TOKEN_FILE`` with owner-only + permissions. Every request must carry this token or it is refused. + """ + env_token = os.environ.get("GRAMPS_AGENT_TOKEN") + if env_token: + return env_token + try: + with open(TOKEN_FILE, "r", encoding="utf-8") as handle: + existing = handle.read().strip() + if existing: + return existing + except OSError: + pass + token = secrets.token_hex(32) + with open(TOKEN_FILE, "w", encoding="utf-8") as handle: + handle.write(token) + try: + os.chmod(TOKEN_FILE, stat.S_IRUSR | stat.S_IWUSR) + except OSError: + pass + return token + + + +def _safe_repr(value): + """ + Return a length-bounded repr of a value, never raising. + """ + try: + text = repr(value) + except Exception: + try: + text = "" % type(value).__name__ + except Exception: + text = "" + if len(text) > MAX_REPR: + text = text[:MAX_REPR] + "... [truncated]" + return text + + +# ------------------------------------------------------------------------- +# +# AgentBridge +# +# ------------------------------------------------------------------------- +class AgentBridge(Gramplet): + """ + Poll a control directory and execute agent requests on the main loop. + """ + + def init(self): + """ + Set up the control directory and start the polling timer. + """ + self._ns = None + self._timer_id = None + self._served = 0 + self._token = None + self.gui.set_text(_("Agent Bridge starting...")) + try: + os.makedirs(REQ_DIR, exist_ok=True) + os.makedirs(RESP_DIR, exist_ok=True) + self._token = _ensure_token() + except OSError as err: + self.gui.set_text(_("Agent Bridge error: %s") % err) + LOG.error("Could not initialize control dir: %s", err) + return + self._start() + self._set_status() + + def _start(self): + """ + Start the GLib poll timer if it is not already running. + """ + if self._timer_id is None: + self._timer_id = GLib.timeout_add(POLL_MS, self._poll) + + def _set_status(self): + """ + Update the gramplet status text. + """ + db_name = _("(no tree)") + try: + if self.dbstate.is_open(): + db_name = self.dbstate.db.get_dbname() + except Exception: + pass + self.gui.set_text( + _("Agent Bridge active.\n") + + _("Watching: %s\n") % REQ_DIR + + _("Tree: %s\n") % db_name + + _("Requests served: %d") % self._served + ) + + def main(self): + """ + Refresh status on update events. No periodic work is done here. + """ + self._set_status() + + def _namespace(self): + """ + Return the persistent exec namespace, refreshing live references. + """ + if self._ns is None: + import gramps.gen.lib as gramps_lib + + self._ns = { + "__name__": "agent_bridge", + "gramps_lib": gramps_lib, + "bridge": self, + } + # Keep live handles fresh on every call -- the tree can change. + self._ns["dbstate"] = self.dbstate + self._ns["uistate"] = self.uistate + self._ns["gui"] = self.gui + self._ns["db"] = self.dbstate.db if self.dbstate.is_open() else None + return self._ns + + def _poll(self): + """ + Process any pending request files. Always returns True to keep polling. + """ + try: + names = sorted( + name for name in os.listdir(REQ_DIR) if name.endswith(".req.json") + ) + except FileNotFoundError: + return True + for name in names: + path = os.path.join(REQ_DIR, name) + try: + with open(path, "r", encoding="utf-8") as handle: + req = json.load(handle) + except (ValueError, OSError): + # File may still be partly written; leave it for the next tick. + continue + try: + os.remove(path) + except OSError: + pass + rid = req.get("id") or name[: -len(".req.json")] + resp = self._handle(req) + self._served += 1 + self._write_response(rid, resp) + self._set_status() + return True + + def _handle(self, req): + """ + Dispatch a single request to the matching action handler. + + Requests must carry the shared secret token or they are refused without + executing anything. + """ + if not self._token or not hmac.compare_digest( + str(req.get("token") or ""), self._token + ): + return {"ok": False, "error": "unauthorized: missing or invalid token"} + action = req.get("action", "eval") + try: + if action == "ping": + return {"ok": True, "pong": True, "served": self._served} + if action == "eval": + return self._do_eval(req.get("code", "")) + if action == "install_plugin": + return self._do_install(req) + return {"ok": False, "error": "unknown action: %r" % action} + except Exception: + return {"ok": False, "error": traceback.format_exc()} + + def _do_eval(self, code): + """ + Execute submitted code in the persistent namespace. + + Captures stdout/stderr. If the code assigns a variable named ``result`` + its repr is returned. + """ + namespace = self._namespace() + namespace["result"] = None + stream = io.StringIO() + try: + compiled = compile(code, "", "exec") + with contextlib.redirect_stdout(stream), contextlib.redirect_stderr( + stream + ): + exec(compiled, namespace) + return { + "ok": True, + "stdout": stream.getvalue(), + "result": _safe_repr(namespace.get("result")), + } + except Exception: + return { + "ok": False, + "stdout": stream.getvalue(), + "error": traceback.format_exc(), + } + + def _do_install(self, req): + """ + Write one or more plugin files to the user plugin directory and reload. + + Expects ``req['files']`` as a mapping of relative path -> file contents, + and an optional ``req['name']`` used as the containing directory. + """ + from gramps.gen.const import USER_PLUGINS + from gramps.gen.plug import BasePluginManager + + files = req.get("files") or {} + if not isinstance(files, dict) or not files: + return {"ok": False, "error": "install_plugin needs a 'files' mapping"} + subdir = req.get("name", "agent_plugin") + target = os.path.join(USER_PLUGINS, subdir) + os.makedirs(target, exist_ok=True) + written = [] + for relpath, content in files.items(): + dest = os.path.join(target, relpath) + os.makedirs(os.path.dirname(dest) or target, exist_ok=True) + with open(dest, "w", encoding="utf-8") as handle: + handle.write(content) + written.append(dest) + pmgr = BasePluginManager.get_instance() + # Scan the freshly written directory so a brand-new plugin registers. + pmgr.reg_plugins(target, self.dbstate, self.uistate, rescan=True) + return {"ok": True, "written": written, "dir": target} + + def _write_response(self, rid, resp): + """ + Write a response file atomically (temp file then rename). + """ + resp.setdefault("ok", True) + resp["id"] = rid + resp["ts"] = time.time() + tmp = os.path.join(RESP_DIR, "%s.part" % rid) + final = os.path.join(RESP_DIR, "%s.resp.json" % rid) + try: + with open(tmp, "w", encoding="utf-8") as handle: + json.dump(resp, handle) + os.replace(tmp, final) + except OSError as err: + LOG.error("Could not write response %s: %s", rid, err) + + def on_save(self): + """ + Stop the poll timer when the gramplet is disposed. + """ + if self._timer_id is not None: + GLib.source_remove(self._timer_id) + self._timer_id = None diff --git a/AgentBridge/MANIFEST b/AgentBridge/MANIFEST new file mode 100644 index 000000000..8eea94deb --- /dev/null +++ b/AgentBridge/MANIFEST @@ -0,0 +1 @@ +AgentBridge/README.md diff --git a/AgentBridge/README.md b/AgentBridge/README.md new file mode 100644 index 000000000..1acbd588d --- /dev/null +++ b/AgentBridge/README.md @@ -0,0 +1,119 @@ +# Agent Bridge + +Agent Bridge embeds a control bridge inside a running Gramps session so an AI +agent can drive the live application — read and modify the family tree, operate +the user interface, run reports, and create and load new plugins on the fly. + +It ships an **MCP server**, so any [Model Context +Protocol](https://modelcontextprotocol.io) client (Claude, and other AI agents) +can drive Gramps through standard tools with no custom glue. + +> ⚠️ **Security**: this addon executes arbitrary Python inside Gramps with your +> user privileges. It listens on **no network port** — control happens purely +> through files under `~/.gramps_agent` — so the trust boundary is your user +> account. Only enable it on a machine you control, and remove the gramplet +> when you are done. +> +> As defense-in-depth, every request must carry a shared secret **token**. The +> gramplet generates it on first run at `~/.gramps_agent/token` (owner-only +> permissions) and refuses any request without it. The bundled MCP server and +> `agent_send.py` read the same file automatically, so on a normal single-user +> machine this is transparent. To use a fixed token (e.g. a synced control +> directory), set `GRAMPS_AGENT_TOKEN` in the environment of both Gramps and +> the MCP server. + +## Architecture + +``` + AI client (Claude / any MCP agent) + │ Model Context Protocol (stdio) + ▼ + gramps_mcp_server.py (a normal Python process) + │ watched control dir (~/.gramps_agent) + ▼ + Agent Bridge gramplet (inside Gramps; GLib poller on the GTK main thread) + │ + ▼ + live Gramps: database, UI state, plugin system +``` + +The gramplet polls the control directory **on the GTK main thread**, so injected +code can safely touch both the database and the GUI. The exec namespace is +persistent — names defined in one call survive to the next — making it a true +REPL. The MCP server is only a protocol adapter; it needs the `mcp` package and +file access, not Gramps' bundled interpreter. + +## Install the gramplet + +1. Copy the `AgentBridge` folder into your Gramps user plugin directory + (e.g. `~/.gramps/gramps60/plugins/` on Linux, or + `%APPDATA%\gramps\gramps60\plugins\` on Windows), or install it from the + Gramps Addon Manager once published. +2. Restart Gramps. +3. On the **Dashboard**, right-click a gramplet bar → **Add a gramplet** → + **Agent Bridge**. It should display *"Agent Bridge active"*. It persists in + your Dashboard layout for future sessions. + +## Wire up the MCP server + +In the Python environment that runs your AI client (not Gramps' bundled one): + +```bash +pip install "mcp[cli]" +``` + +Register the server with Claude Code: + +```bash +claude mcp add gramps -- python /path/to/AgentBridge/gramps_mcp_server.py +``` + +…or add it to a project `.mcp.json`: + +```json +{ + "mcpServers": { + "gramps": { + "command": "python", + "args": ["/path/to/AgentBridge/gramps_mcp_server.py"] + } + } +} +``` + +If your control directory is not the default, set `GRAMPS_AGENT_DIR` in the +server's environment so both sides agree. + +## Tools exposed over MCP + +| Tool | What it does | +|------|--------------| +| `gramps_status` | Confirm the bridge is reachable; report requests served. | +| `gramps_eval` | Run Python in the live process (`db`, `dbstate`, `uistate`, `gui`, `gramps_lib`, `bridge` pre-bound; assign `result`). | +| `gramps_install_plugin` | Write a plugin into the user plugin dir and hot-reload it. | +| `gramps_people_count` | Number of people in the open tree. | +| `gramps_search_people` | Substring search by name; returns `{gramps_id, name}`. | +| `gramps_active_person` | The currently selected person. | +| `gramps_set_active_person` | Navigate Gramps to a person by Gramps ID. | + +`gramps_eval` is the universal primitive; the others are convenience wrappers an +agent can reach for directly. + +## Without MCP (debugging) + +`agent_send.py` is a low-level CLI that talks to the bridge directly: + +```bash +python agent_send.py ping +python agent_send.py eval -c "result = db.get_number_of_people()" +``` + +## Uninstall / disable + +Remove the **Agent Bridge** gramplet from the Dashboard, or delete the +`AgentBridge` plugin folder, then restart Gramps. You may also delete +`~/.gramps_agent`. + +## Contact + +Brian Caudill — brian.m.caudill@gmail.com diff --git a/AgentBridge/agent_send.py b/AgentBridge/agent_send.py new file mode 100644 index 000000000..78602400b --- /dev/null +++ b/AgentBridge/agent_send.py @@ -0,0 +1,161 @@ +# +# Gramps - a GTK+/GNOME based genealogy program +# +# Copyright (C) 2026 Brian Caudill +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along +# with this program; if not, see . +# + +""" +Low-level command line client for the Gramps Agent Bridge gramplet. + +This is a debugging aid that bypasses MCP and talks to the bridge directly. +For normal AI-driven use, prefer the MCP server (gramps_mcp_server.py). + +Usage: + python agent_send.py ping + python agent_send.py eval -c "result = db.get_number_of_people()" + python agent_send.py eval -f snippet.py + echo "result = 1 + 1" | python agent_send.py eval +""" +import os +import sys +import json +import time +import uuid +import argparse + +CONTROL = os.environ.get( + "GRAMPS_AGENT_DIR", os.path.join(os.path.expanduser("~"), ".gramps_agent") +) +REQ = os.path.join(CONTROL, "requests") +RESP = os.path.join(CONTROL, "responses") +TOKEN_FILE = os.path.join(CONTROL, "token") + + +def read_token(): + """Return the shared secret expected by the bridge, or '' if not found.""" + env_token = os.environ.get("GRAMPS_AGENT_TOKEN") + if env_token: + return env_token + try: + with open(TOKEN_FILE, "r", encoding="utf-8") as handle: + return handle.read().strip() + except OSError: + return "" + + +def send(action, code=None, files=None, name=None, timeout=60): + """Write a request and block until the response arrives or timeout.""" + os.makedirs(REQ, exist_ok=True) + os.makedirs(RESP, exist_ok=True) + rid = uuid.uuid4().hex[:12] + req = {"id": rid, "action": action, "token": read_token()} + if code is not None: + req["code"] = code + if files is not None: + req["files"] = files + if name is not None: + req["name"] = name + tmp = os.path.join(REQ, rid + ".req.json.part") + final = os.path.join(REQ, rid + ".req.json") + with open(tmp, "w", encoding="utf-8") as handle: + json.dump(req, handle) + os.replace(tmp, final) + + respfile = os.path.join(RESP, rid + ".resp.json") + deadline = time.time() + timeout + while time.time() < deadline: + if os.path.exists(respfile): + try: + with open(respfile, "r", encoding="utf-8") as handle: + resp = json.load(handle) + except (ValueError, OSError): + time.sleep(0.1) + continue + try: + os.remove(respfile) + except OSError: + pass + return resp + time.sleep(0.15) + return { + "ok": False, + "error": "timeout after %ss -- is the Agent Bridge gramplet added " + "and Gramps running?" % timeout, + } + + +def main(): + parser = argparse.ArgumentParser(description="Drive the Gramps Agent Bridge") + parser.add_argument("action", choices=["ping", "eval", "install"]) + parser.add_argument("-c", "--code", help="inline code for eval") + parser.add_argument("-f", "--file", help="read eval code from this file") + parser.add_argument("-n", "--name", help="plugin dir name for install") + parser.add_argument( + "-F", + "--plugin-file", + action="append", + default=[], + metavar="DEST=SRCPATH", + help="install file mapping, repeatable", + ) + parser.add_argument("-t", "--timeout", type=float, default=60) + parser.add_argument("--raw", action="store_true", help="print raw JSON only") + args = parser.parse_args() + + code = None + files = None + if args.action == "eval": + if args.code is not None: + code = args.code + elif args.file: + with open(args.file, "r", encoding="utf-8") as handle: + code = handle.read() + else: + code = sys.stdin.read() + elif args.action == "install": + files = {} + for mapping in args.plugin_file: + dest, _, srcpath = mapping.partition("=") + with open(srcpath, "r", encoding="utf-8") as handle: + files[dest] = handle.read() + + action = "install_plugin" if args.action == "install" else args.action + resp = send( + action, code=code, files=files, name=args.name, timeout=args.timeout + ) + + if args.raw: + print(json.dumps(resp, indent=2)) + return 0 if resp.get("ok") else 1 + + print("ok:", resp.get("ok")) + if resp.get("stdout"): + print("--- stdout ---") + print(resp["stdout"], end="" if resp["stdout"].endswith("\n") else "\n") + if "result" in resp and resp["result"] not in (None, "None"): + print("--- result ---") + print(resp["result"]) + if resp.get("error"): + print("--- error ---") + print(resp["error"], end="" if resp["error"].endswith("\n") else "\n") + if resp.get("written"): + print("--- installed ---") + print("\n".join(resp["written"])) + return 0 if resp.get("ok") else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/AgentBridge/gramps_mcp_server.py b/AgentBridge/gramps_mcp_server.py new file mode 100644 index 000000000..89db2671f --- /dev/null +++ b/AgentBridge/gramps_mcp_server.py @@ -0,0 +1,235 @@ +# +# Gramps - a GTK+/GNOME based genealogy program +# +# Copyright (C) 2026 Brian Caudill +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License along +# with this program; if not, see . +# + +""" +MCP server for the Gramps Agent Bridge. + +Exposes the live Gramps application to any MCP-capable AI as a set of tools. +This process speaks the Model Context Protocol over stdio to the AI client and +forwards each call to the Agent Bridge gramplet running inside Gramps, using a +shared control directory (default ``~/.gramps_agent``). + +Requirements (in the Python that launches this server, NOT Gramps' bundled +interpreter): + + pip install "mcp[cli]" + +Register with an MCP client, e.g. Claude Code: + + claude mcp add gramps -- python /path/to/gramps_mcp_server.py + +or add to a project ``.mcp.json``: + + { + "mcpServers": { + "gramps": { "command": "python", + "args": ["/path/to/gramps_mcp_server.py"] } + } + } + +The Agent Bridge gramplet must be added to the Gramps Dashboard for these tools +to do anything; otherwise calls time out with a helpful message. +""" +import os +import json +import time +import uuid + +from mcp.server.fastmcp import FastMCP + +CONTROL_DIR = os.environ.get( + "GRAMPS_AGENT_DIR", os.path.join(os.path.expanduser("~"), ".gramps_agent") +) +REQ_DIR = os.path.join(CONTROL_DIR, "requests") +RESP_DIR = os.path.join(CONTROL_DIR, "responses") +TOKEN_FILE = os.path.join(CONTROL_DIR, "token") +DEFAULT_TIMEOUT = float(os.environ.get("GRAMPS_AGENT_TIMEOUT", "60")) + +mcp = FastMCP("gramps") + + +def _read_token(): + """Return the shared secret expected by the bridge, or '' if not found. + + Read fresh each call so a token the gramplet generates after this server + starts is still picked up.""" + env_token = os.environ.get("GRAMPS_AGENT_TOKEN") + if env_token: + return env_token + try: + with open(TOKEN_FILE, "r", encoding="utf-8") as handle: + return handle.read().strip() + except OSError: + return "" + + +# ------------------------------------------------------------------------- +# +# Transport: write a request file, wait for the response file +# +# ------------------------------------------------------------------------- +def _call(action, timeout=DEFAULT_TIMEOUT, **payload): + """Send one request to the bridge and block for its response.""" + os.makedirs(REQ_DIR, exist_ok=True) + os.makedirs(RESP_DIR, exist_ok=True) + rid = uuid.uuid4().hex[:12] + request = {"id": rid, "action": action, "token": _read_token()} + request.update(payload) + + tmp = os.path.join(REQ_DIR, rid + ".req.json.part") + final = os.path.join(REQ_DIR, rid + ".req.json") + with open(tmp, "w", encoding="utf-8") as handle: + json.dump(request, handle) + os.replace(tmp, final) + + respfile = os.path.join(RESP_DIR, rid + ".resp.json") + deadline = time.time() + timeout + while time.time() < deadline: + if os.path.exists(respfile): + try: + with open(respfile, "r", encoding="utf-8") as handle: + resp = json.load(handle) + except (ValueError, OSError): + time.sleep(0.1) + continue + try: + os.remove(respfile) + except OSError: + pass + return resp + time.sleep(0.1) + return { + "ok": False, + "error": ( + "Timed out after %ss. Is Gramps running with the 'Agent Bridge' " + "gramplet added to the Dashboard? Control dir: %s" % (timeout, CONTROL_DIR) + ), + } + + +# ------------------------------------------------------------------------- +# +# Core tools +# +# ------------------------------------------------------------------------- +@mcp.tool() +def gramps_status() -> dict: + """Check whether the Gramps Agent Bridge is reachable and report how many + requests it has served. Use this first to confirm the connection.""" + return _call("ping", timeout=10) + + +@mcp.tool() +def gramps_eval(code: str, timeout: float = DEFAULT_TIMEOUT) -> dict: + """Run Python code inside the live Gramps process and return its output. + + The code runs on the GTK main thread in a persistent namespace, so it can + safely read or modify the family tree and drive the user interface, and + names defined in one call remain available to later calls. These names are + pre-bound: ``dbstate`` (the DbState), ``db`` (the open database or None), + ``uistate`` (the UIState), ``gui`` (the gramplet view), ``gramps_lib`` + (gramps.gen.lib), and ``bridge`` (the gramplet itself). + + Assign to a variable named ``result`` to return a value; anything printed + is captured as ``stdout``. Example: + result = db.get_number_of_people() + Returns a dict with ``ok`` and either ``stdout``/``result`` or ``error``.""" + return _call("eval", code=code, timeout=timeout) + + +@mcp.tool() +def gramps_install_plugin(name: str, files: dict, timeout: float = 30) -> dict: + """Write a Gramps plugin into the user plugin directory and hot-reload it. + + ``name`` is the plugin subdirectory; ``files`` maps relative file paths to + their text contents (include a ``*.gpr.py`` registration file plus the + module). After this the plugin is registered and can be run via gramps_eval. + Returns the list of written paths.""" + return _call("install_plugin", name=name, files=files, timeout=timeout) + + +# ------------------------------------------------------------------------- +# +# Convenience tools (thin wrappers over gramps_eval) +# +# ------------------------------------------------------------------------- +@mcp.tool() +def gramps_people_count() -> dict: + """Return the number of people in the currently open family tree.""" + return _call( + "eval", + code="result = db.get_number_of_people() if db else 'no tree open'", + timeout=20, + ) + + +@mcp.tool() +def gramps_search_people(text: str, limit: int = 25) -> dict: + """Search people by surname or given name (case-insensitive substring). + + Returns up to ``limit`` matches as a list of {gramps_id, name} dicts.""" + code = ( + "matches = []\n" + "needle = %r.lower()\n" + "if db:\n" + " for person in db.iter_people():\n" + " name = person.get_primary_name().get_name()\n" + " if needle in name.lower():\n" + " matches.append({'gramps_id': person.get_gramps_id(),\n" + " 'name': name})\n" + " if len(matches) >= %d:\n" + " break\n" + "result = matches\n" % (text, int(limit)) + ) + return _call("eval", code=code, timeout=60) + + +@mcp.tool() +def gramps_active_person() -> dict: + """Return the gramps_id and name of the currently active (selected) person, + or a note if none is active.""" + code = ( + "handle = uistate.get_active('Person') if uistate else None\n" + "if handle and db:\n" + " p = db.get_person_from_handle(handle)\n" + " result = {'gramps_id': p.get_gramps_id(),\n" + " 'name': p.get_primary_name().get_name()}\n" + "else:\n" + " result = 'no active person'\n" + ) + return _call("eval", code=code, timeout=20) + + +@mcp.tool() +def gramps_set_active_person(gramps_id: str) -> dict: + """Make the person with the given Gramps ID the active person, which drives + navigation across all Gramps views. Returns the activated person's name.""" + code = ( + "p = db.get_person_from_gramps_id(%r) if db else None\n" + "if p is None:\n" + " result = 'no such person: %s'\n" + "else:\n" + " uistate.set_active(p.get_handle(), 'Person')\n" + " result = p.get_primary_name().get_name()\n" % (gramps_id, gramps_id) + ) + return _call("eval", code=code, timeout=20) + + +if __name__ == "__main__": + mcp.run() diff --git a/AgentBridge/po/template.pot b/AgentBridge/po/template.pot new file mode 100644 index 000000000..eed33b92b --- /dev/null +++ b/AgentBridge/po/template.pot @@ -0,0 +1,62 @@ +# SOME DESCRIPTIVE TITLE. +# Copyright (C) YEAR THE PACKAGE'S COPYRIGHT HOLDER +# This file is distributed under the same license as the PACKAGE package. +# FIRST AUTHOR , YEAR. +# +#, fuzzy +msgid "" +msgstr "" +"Project-Id-Version: PACKAGE VERSION\n" +"Report-Msgid-Bugs-To: \n" +"POT-Creation-Date: 2026-06-01 00:00+0000\n" +"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n" +"Last-Translator: FULL NAME \n" +"Language-Team: LANGUAGE \n" +"Language: \n" +"MIME-Version: 1.0\n" +"Content-Type: text/plain; charset=UTF-8\n" +"Content-Transfer-Encoding: 8bit\n" + +#: AgentBridge/AgentBridge.gpr.py:28 AgentBridge/AgentBridge.gpr.py:41 +msgid "Agent Bridge" +msgstr "" + +#: AgentBridge/AgentBridge.gpr.py:29 +msgid "" +"Embeds a control bridge in Gramps so an AI agent can drive the live " +"application through a watched directory or an MCP server. Executes submitted " +"Python on the GTK main thread. For developers and power users; runs arbitrary " +"code at your privileges." +msgstr "" + +#: AgentBridge/AgentBridge.py:122 +msgid "Agent Bridge starting..." +msgstr "" + +#: AgentBridge/AgentBridge.py:127 +#, python-format +msgid "Agent Bridge error: %s" +msgstr "" + +#: AgentBridge/AgentBridge.py:144 +msgid "(no tree)" +msgstr "" + +#: AgentBridge/AgentBridge.py:151 +msgid "Agent Bridge active.\n" +msgstr "" + +#: AgentBridge/AgentBridge.py:152 +#, python-format +msgid "Watching: %s\n" +msgstr "" + +#: AgentBridge/AgentBridge.py:153 +#, python-format +msgid "Tree: %s\n" +msgstr "" + +#: AgentBridge/AgentBridge.py:154 +#, python-format +msgid "Requests served: %d" +msgstr ""