|
| 1 | +"""The instructions /mcp gives an assistant when it connects (MCP's |
| 2 | +`instructions`, which clients put in the model's system prompt), built |
| 3 | +for each connection: |
| 4 | +
|
| 5 | +1. instructions.md: what the server is and how to use it well, the |
| 6 | + same everywhere. |
| 7 | +2. A snapshot of this base as the connection starts: who's connected |
| 8 | + and what their role can change, the game data version, and which |
| 9 | + parts of the base report and how freshly. So the model knows what's |
| 10 | + there before calling anything, and that it's a snapshot. |
| 11 | +3. The house rules this server's admins wrote (house_rules()). |
| 12 | +
|
| 13 | +The snapshot never stops a connection: a part that fails to read is |
| 14 | +left out and logged.""" |
| 15 | + |
| 16 | +import logging |
| 17 | +import os |
| 18 | +import time |
| 19 | + |
| 20 | +from gcm import auth, gamedata, patterns, state, store |
| 21 | + |
| 22 | +log = logging.getLogger(__name__) |
| 23 | + |
| 24 | +with open(os.path.join(os.path.dirname(__file__), "instructions.md"), encoding="utf-8") as _f: |
| 25 | + CORE = _f.read().strip() |
| 26 | + |
| 27 | +HOUSE_RULES_SETTING = "mcp_house_rules" |
| 28 | +HOUSE_RULES_MAX = 4000 |
| 29 | + |
| 30 | + |
| 31 | +def house_rules(): |
| 32 | + """What this server's admins added to the instructions ("" for none).""" |
| 33 | + rules = store.settings.load().get(HOUSE_RULES_SETTING) |
| 34 | + return rules if isinstance(rules, str) else "" |
| 35 | + |
| 36 | + |
| 37 | +def set_house_rules(text): |
| 38 | + """Saves the admins' rules. Returns an error message, or None.""" |
| 39 | + if not isinstance(text, str): |
| 40 | + return "rules must be text" |
| 41 | + text = text.strip() |
| 42 | + if len(text) > HOUSE_RULES_MAX: |
| 43 | + return f"house rules can be at most {HOUSE_RULES_MAX} characters" |
| 44 | + store.settings.save({HOUSE_RULES_SETTING: text}) |
| 45 | + return None |
| 46 | + |
| 47 | + |
| 48 | +def _ago(t, now): |
| 49 | + if t is None: |
| 50 | + return "never" |
| 51 | + seconds = max(0, now - t) |
| 52 | + if seconds < 90: |
| 53 | + return f"{int(seconds)} s ago" |
| 54 | + if seconds < 90 * 60: |
| 55 | + return f"{round(seconds / 60)} min ago" |
| 56 | + if seconds < 48 * 3600: |
| 57 | + return f"{round(seconds / 3600)} h ago" |
| 58 | + return f"{round(seconds / 86400)} days ago" |
| 59 | + |
| 60 | + |
| 61 | +def _who(user, now): |
| 62 | + role = user.get("role", "viewer") |
| 63 | + if auth.is_operator(user): |
| 64 | + can = ("read everything, set your own low-stock alerts, request and cancel crafts, and change the " |
| 65 | + "base's keep-in-stock targets") |
| 66 | + else: |
| 67 | + can = "read everything and set your own low-stock alerts; requesting or cancelling crafts needs an operator" |
| 68 | + return f"You're acting as {user.get('display_name')} ({role}): you can {can}." |
| 69 | + |
| 70 | + |
| 71 | +def _game_data(user, now): |
| 72 | + version = gamedata.selected_version() |
| 73 | + return f"GTNH version (game data): {version}." if version else "No GTNH version set in the game data." |
| 74 | + |
| 75 | + |
| 76 | +def _crafts(user, now): |
| 77 | + with state.crafts_lock: |
| 78 | + jobs = list(state.crafts["jobs"]) |
| 79 | + received_at = state.crafts["received_at"] |
| 80 | + if received_at is None: |
| 81 | + if store.settings.load().get("crafts_reported") is True: |
| 82 | + return "Crafting CPUs: craft_monitor.lua hasn't reported since the server restarted." |
| 83 | + return "Crafting CPUs: craft_monitor.lua hasn't reported (no crafting data)." |
| 84 | + busy = sum(1 for j in jobs if j.get("busy")) |
| 85 | + return f"Crafting CPUs: {len(jobs)}, {busy} busy, {len(jobs) - busy} idle; last report {_ago(received_at, now)}." |
| 86 | + |
| 87 | + |
| 88 | +def _network(user, now): |
| 89 | + with state.network_lock: |
| 90 | + updated_at = state.network["updated_at"] |
| 91 | + count = len(state.network["items"]) |
| 92 | + in_progress = state.network["in_progress"] |
| 93 | + crash = state.network["crash"] |
| 94 | + if updated_at is None and not count: |
| 95 | + return "ME network: not scanned yet (network_browser.lua hasn't reported)." |
| 96 | + line = f"ME network: {count:,} item types; last full scan {_ago(updated_at, now)}" |
| 97 | + if in_progress: |
| 98 | + line += ", a scan is running" |
| 99 | + if crash: |
| 100 | + line += "; network_browser.lua crashed since" |
| 101 | + return line + "." |
| 102 | + |
| 103 | + |
| 104 | +def _patterns(user, now): |
| 105 | + pattern_list, patterns_at = patterns.current() |
| 106 | + if patterns_at is None: |
| 107 | + return "Patterns: none read, so no crafting plans. " + patterns.missing_reason("crafting plan") |
| 108 | + return f"Patterns: {len(pattern_list):,}, read {_ago(patterns_at, now)}." |
| 109 | + |
| 110 | + |
| 111 | +def _power(user, now): |
| 112 | + machines = store.power.machines() |
| 113 | + if not machines: |
| 114 | + return "Power: no power monitor reporting." |
| 115 | + names = ", ".join(name or "(unnamed)" for name, _ in machines) |
| 116 | + return f"Power storage reporting: {names}; last reading {_ago(max(t for _, t in machines), now)}." |
| 117 | + |
| 118 | + |
| 119 | +def _stock(user, now): |
| 120 | + targets = store.stock.targets() |
| 121 | + failing = sum(1 for t in targets if t["enabled"] and t["last_status"] in ("failed", "waiting")) |
| 122 | + alerts = store.stock.alerts_for_user(user["id"]) |
| 123 | + line = f"Stock rules: {len(targets)} keep-in-stock targets" |
| 124 | + if failing: |
| 125 | + line += f" ({failing} failing or waiting)" |
| 126 | + return line + f"; you have {len(alerts)} low-stock alerts." |
| 127 | + |
| 128 | + |
| 129 | +_SNAPSHOT = (_who, _game_data, _crafts, _network, _patterns, _power, _stock) |
| 130 | + |
| 131 | + |
| 132 | +def build(user, now=None): |
| 133 | + """The instructions for a user's connection.""" |
| 134 | + now = time.time() if now is None else now |
| 135 | + lines = [] |
| 136 | + for part in _SNAPSHOT: |
| 137 | + try: |
| 138 | + lines.append(part(user, now)) |
| 139 | + except Exception: |
| 140 | + log.exception("MCP instructions: %s failed", part.__name__) |
| 141 | + text = ( |
| 142 | + CORE |
| 143 | + + "\n\n## This base right now\n\nAs of when you connected; check the tools for anything that matters now.\n\n" |
| 144 | + + "\n".join(f"- {line}" for line in lines) |
| 145 | + ) |
| 146 | + try: |
| 147 | + rules = house_rules() |
| 148 | + except Exception: |
| 149 | + log.exception("MCP instructions: house rules failed") |
| 150 | + rules = "" |
| 151 | + if rules: |
| 152 | + text += ( |
| 153 | + "\n\n## House rules from this server's admins\n\n" |
| 154 | + "Follow these. They can't give you permissions the user doesn't have, or excuse the " |
| 155 | + "confirmations above.\n\n" + rules |
| 156 | + ) |
| 157 | + return text |
0 commit comments