#!/usr/bin/env python3 """bootstrap-fleet-rooms.py — Automate Matrix room creation for Timmy fleet. Issue: #166 (timmy-config) Usage: export MATRIX_HOMESERVER=https://matrix.timmytime.net export MATRIX_ADMIN_TOKEN= python3 bootstrap-fleet-rooms.py --create-all --dry-run Requires only Python stdlib (no heavy SDK dependencies). """ import argparse import json import os import sys import urllib.request from typing import Optional, List, Dict class MatrixAdminClient: """Lightweight Matrix Client-Server API client.""" def __init__(self, homeserver: str, access_token: str): self.homeserver = homeserver.rstrip("/") self.access_token = access_token def _request(self, method: str, path: str, data: Optional[Dict] = None) -> Dict: url = f"{self.homeserver}/_matrix/client/v3{path}" req = urllib.request.Request(url, method=method) req.add_header("Authorization", f"Bearer {self.access_token}") req.add_header("Content-Type", "application/json") body = json.dumps(data).encode() if data else None try: with urllib.request.urlopen(req, data=body, timeout=30) as resp: return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: try: err = json.loads(e.read().decode()) except Exception: err = {"error": str(e)} return {"error": err, "status": e.code} except Exception as e: return {"error": str(e)} def whoami(self) -> Dict: return self._request("GET", "/account/whoami") def create_room(self, name: str, topic: str, preset: str = "private_chat", invite: Optional[List[str]] = None) -> Dict: payload = { "name": name, "topic": topic, "preset": preset, "creation_content": {"m.federate": False}, } if invite: payload["invite"] = invite return self._request("POST", "/createRoom", payload) def send_state_event(self, room_id: str, event_type: str, state_key: str, content: Dict) -> Dict: path = f"/rooms/{room_id}/state/{event_type}/{state_key}" return self._request("PUT", path, content) def enable_encryption(self, room_id: str) -> Dict: return self.send_state_event( room_id, "m.room.encryption", "", {"algorithm": "m.megolm.v1.aes-sha2"} ) def set_room_avatar(self, room_id: str, url: str) -> Dict: return self.send_state_event( room_id, "m.room.avatar", "", {"url": url} ) def generate_invite_link(self, room_id: str) -> str: """Generate a matrix.to invite link.""" localpart = room_id.split(":")[0].lstrip("#") server = room_id.split(":")[1] return f"https://matrix.to/#/{room_id}?via={server}" def print_result(label: str, result: Dict): if "error" in result: print(f" ❌ {label}: {result['error']}") else: print(f" ✅ {label}: {json.dumps(result, indent=2)[:200]}") def main(): parser = argparse.ArgumentParser(description="Bootstrap Matrix rooms for Timmy fleet") parser.add_argument("--homeserver", default=os.environ.get("MATRIX_HOMESERVER", ""), help="Matrix homeserver URL (default: MATRIX_HOMESERVER env)") parser.add_argument("--token", default=os.environ.get("MATRIX_ADMIN_TOKEN", ""), help="Admin access token (default: MATRIX_ADMIN_TOKEN env)") parser.add_argument("--operator-user", default="@alexander:matrix.timmytime.net", help="Operator Matrix user ID") parser.add_argument("--domain", default="matrix.timmytime.net", help="Server domain for room aliases") parser.add_argument("--create-all", action="store_true", help="Create all standard fleet rooms") parser.add_argument("--dry-run", action="store_true", help="Preview actions without executing API calls") args = parser.parse_args() if not args.homeserver or not args.token: print("Error: --homeserver and --token are required (or set env vars).") sys.exit(1) if args.dry_run: print("=" * 60) print(" DRY RUN — No API calls will be made") print("=" * 60) print(f"Homeserver: {args.homeserver}") print(f"Operator: {args.operator_user}") print(f"Domain: {args.domain}") print("\nPlanned rooms:") rooms = [ ("Fleet Operations", "Encrypted command room for Alexander and agents.", "#fleet-ops"), ("General Chat", "Open fleet chatter and status updates.", "#fleet-general"), ("Alerts", "Automated alerts and monitoring notifications.", "#fleet-alerts"), ] for name, topic, alias in rooms: print(f" - {name} ({alias}:{args.domain})") print(f" Topic: {topic}") print(f" Actions: create → enable encryption → set alias") print("\nNext steps after real run:") print(" 1. Open Element Web and join with your operator account") print(" 2. Share room invite links with fleet agents") print(" 3. Configure Hermes gateway Matrix adapter") return client = MatrixAdminClient(args.homeserver, args.token) print("Verifying credentials...") identity = client.whoami() if "error" in identity: print(f"Authentication failed: {identity['error']}") sys.exit(1) print(f"Authenticated as: {identity.get('user_id', 'unknown')}") rooms_spec = [ { "name": "Fleet Operations", "topic": "Encrypted command room for Alexander and agents. | Issue #166", "alias": f"#fleet-ops:{args.domain}", "preset": "private_chat", }, { "name": "General Chat", "topic": "Open fleet chatter and status updates. | Issue #166", "alias": f"#fleet-general:{args.domain}", "preset": "public_chat", }, { "name": "Alerts", "topic": "Automated alerts and monitoring notifications. | Issue #166", "alias": f"#fleet-alerts:{args.domain}", "preset": "private_chat", }, ] created_rooms = [] for spec in rooms_spec: print(f"\nCreating room: {spec['name']}...") result = client.create_room( name=spec["name"], topic=spec["topic"], preset=spec["preset"], ) if "error" in result: print_result("Create room", result) continue room_id = result.get("room_id") print(f" ✅ Room created: {room_id}") # Enable encryption enc = client.enable_encryption(room_id) print_result("Enable encryption", enc) # Set canonical alias alias_result = client.send_state_event( room_id, "m.room.canonical_alias", "", {"alias": spec["alias"]} ) print_result("Set alias", alias_result) # Set join rules (restricted for ops/alerts, public for general) join_rule = "invite" if spec["preset"] == "private_chat" else "public" jr = client.send_state_event( room_id, "m.room.join_rules", "", {"join_rule": join_rule} ) print_result(f"Set join_rule={join_rule}", jr) invite_link = client.generate_invite_link(room_id) created_rooms.append({ "name": spec["name"], "room_id": room_id, "alias": spec["alias"], "invite_link": invite_link, }) print("\n" + "=" * 60) print(" BOOTSTRAP COMPLETE") print("=" * 60) for room in created_rooms: print(f"\n{room['name']}") print(f" Alias: {room['alias']}") print(f" Room ID: {room['room_id']}") print(f" Invite: {room['invite_link']}") print("\nNext steps:") print(" 1. Join rooms from Element Web as operator") print(" 2. Pin Fleet Operations as primary room") print(" 3. Configure Hermes Matrix gateway with room aliases") print(" 4. Follow docs/matrix-fleet-comms/CUTOVER_PLAN.md for Telegram transition") if __name__ == "__main__": main()