feat: bootstrap local Evennia world lane for Timmy (#36)
This commit is contained in:
1
evennia_tools/__init__.py
Normal file
1
evennia_tools/__init__.py
Normal file
@@ -0,0 +1 @@
|
|||||||
|
"""Evennia helper modules for Timmy's persistent world lane."""
|
||||||
62
evennia_tools/layout.py
Normal file
62
evennia_tools/layout.py
Normal file
@@ -0,0 +1,62 @@
|
|||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class RoomSpec:
|
||||||
|
key: str
|
||||||
|
desc: str
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class ExitSpec:
|
||||||
|
source: str
|
||||||
|
key: str
|
||||||
|
destination: str
|
||||||
|
aliases: tuple[str, ...] = ()
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class ObjectSpec:
|
||||||
|
key: str
|
||||||
|
location: str
|
||||||
|
desc: str
|
||||||
|
|
||||||
|
|
||||||
|
ROOMS = (
|
||||||
|
RoomSpec("Gate", "A deliberate threshold into Timmy's world. The air is still here, as if entry itself matters."),
|
||||||
|
RoomSpec("Courtyard", "The central open court of Timmy's place. Paths lead outward to work, memory, prayer, and watchfulness."),
|
||||||
|
RoomSpec("Workshop", "Benches, tools, half-built mechanisms, and active prototypes fill the room. This is where ideas become artifacts."),
|
||||||
|
RoomSpec("Archive", "Shelves hold transcripts, reports, doctrine-bearing documents, and recovered fragments. This room remembers structure."),
|
||||||
|
RoomSpec("Chapel", "A quiet room set apart for prayer, conscience, grief, and right alignment. The tone here is gentle and unhurried."),
|
||||||
|
)
|
||||||
|
|
||||||
|
EXITS = (
|
||||||
|
ExitSpec("Gate", "enter", "Courtyard", ("in", "forward")),
|
||||||
|
ExitSpec("Courtyard", "gate", "Gate", ("out", "threshold")),
|
||||||
|
ExitSpec("Courtyard", "workshop", "Workshop", ("work",)),
|
||||||
|
ExitSpec("Workshop", "courtyard", "Courtyard", ("back", "out")),
|
||||||
|
ExitSpec("Courtyard", "archive", "Archive", ("memory",)),
|
||||||
|
ExitSpec("Archive", "courtyard", "Courtyard", ("back", "out")),
|
||||||
|
ExitSpec("Courtyard", "chapel", "Chapel", ("quiet", "prayer")),
|
||||||
|
ExitSpec("Chapel", "courtyard", "Courtyard", ("back", "out")),
|
||||||
|
)
|
||||||
|
|
||||||
|
OBJECTS = (
|
||||||
|
ObjectSpec("Book of the Soul", "Chapel", "A doctrinal anchor. It is not decorative; it is a reference point."),
|
||||||
|
ObjectSpec("Workbench", "Workshop", "A broad workbench for prototypes, repairs, and experiments not yet stable enough for the world."),
|
||||||
|
ObjectSpec("Map Table", "Courtyard", "A living map of the known place, useful for orientation and future expansion."),
|
||||||
|
ObjectSpec("Prayer Wall", "Chapel", "A place for names, griefs, mercies, and remembered burdens."),
|
||||||
|
ObjectSpec("Memory Shelves", "Archive", "Shelves prepared for durable artifacts that should outlive a single session."),
|
||||||
|
ObjectSpec("Mirror of Sessions", "Archive", "A reflective object meant to bind world continuity to Hermes session history."),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def room_keys() -> tuple[str, ...]:
|
||||||
|
return tuple(room.key for room in ROOMS)
|
||||||
|
|
||||||
|
|
||||||
|
def grouped_exits() -> dict[str, tuple[ExitSpec, ...]]:
|
||||||
|
out: dict[str, list[ExitSpec]] = {}
|
||||||
|
for ex in EXITS:
|
||||||
|
out.setdefault(ex.source, []).append(ex)
|
||||||
|
return {k: tuple(v) for k, v in out.items()}
|
||||||
28
evennia_tools/telemetry.py
Normal file
28
evennia_tools/telemetry.py
Normal file
@@ -0,0 +1,28 @@
|
|||||||
|
import json
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
|
def telemetry_dir(base_dir: str | Path = "~/.timmy/training-data/evennia") -> Path:
|
||||||
|
return Path(base_dir).expanduser()
|
||||||
|
|
||||||
|
|
||||||
|
def event_log_path(session_id: str, base_dir: str | Path = "~/.timmy/training-data/evennia") -> Path:
|
||||||
|
session_id = (session_id or "unbound").strip() or "unbound"
|
||||||
|
day = datetime.now(timezone.utc).strftime("%Y%m%d")
|
||||||
|
return telemetry_dir(base_dir) / day / f"{session_id}.jsonl"
|
||||||
|
|
||||||
|
|
||||||
|
def append_event(session_id: str, event: dict, base_dir: str | Path = "~/.timmy/training-data/evennia") -> Path:
|
||||||
|
path = event_log_path(session_id, base_dir)
|
||||||
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
payload = dict(event)
|
||||||
|
payload.setdefault("timestamp", datetime.now(timezone.utc).isoformat())
|
||||||
|
with path.open("a", encoding="utf-8") as f:
|
||||||
|
f.write(json.dumps(payload, ensure_ascii=False) + "\n")
|
||||||
|
return path
|
||||||
|
|
||||||
|
|
||||||
|
def excerpt(text: str, limit: int = 240) -> str:
|
||||||
|
text = " ".join((text or "").split())
|
||||||
|
return text if len(text) <= limit else text[: limit - 3] + "..."
|
||||||
105
reports/production/2026-03-28-evennia-world-proof.md
Normal file
105
reports/production/2026-03-28-evennia-world-proof.md
Normal file
@@ -0,0 +1,105 @@
|
|||||||
|
# Evennia World Proof — 2026-03-28
|
||||||
|
|
||||||
|
Issue:
|
||||||
|
- #36 Stand up local Evennia world and Hermes control path
|
||||||
|
|
||||||
|
World-state proof summary:
|
||||||
|
|
||||||
|
## Local runtime
|
||||||
|
- Evennia runtime path: `~/.timmy/evennia/timmy_world`
|
||||||
|
- Web surface: `http://127.0.0.1:4001`
|
||||||
|
- Telnet surface: `127.0.0.1:4000`
|
||||||
|
|
||||||
|
## Bootstrap proof
|
||||||
|
Command:
|
||||||
|
- `python3 scripts/evennia/bootstrap_local_evennia.py`
|
||||||
|
|
||||||
|
Key output:
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"provision_stdout": "44 objects imported automatically (use -v 2 for details).\n\nWORLD_OK\nTIMMY_LOCATION Gate\n",
|
||||||
|
"verification": {
|
||||||
|
"status": "Portal: RUNNING (pid 93515)\nServer: RUNNING (pid 93565)",
|
||||||
|
"info": "timmy_world Portal 6.0.0 ... telnet: 4000 ... webserver-proxy: 4001 ...",
|
||||||
|
"web": "200\ntext/html; charset=utf-8"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
## Verifier proof
|
||||||
|
Command:
|
||||||
|
- `python3 scripts/evennia/verify_local_evennia.py`
|
||||||
|
|
||||||
|
Observed output:
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"game_dir_exists": true,
|
||||||
|
"web_status": 200,
|
||||||
|
"web_content_type": "text/html; charset=utf-8",
|
||||||
|
"timmy_home": "Gate",
|
||||||
|
"rooms": {
|
||||||
|
"Gate": true,
|
||||||
|
"Courtyard": true,
|
||||||
|
"Workshop": true,
|
||||||
|
"Archive": true,
|
||||||
|
"Chapel": true
|
||||||
|
},
|
||||||
|
"telnet_roundtrip_excerpt": "You become Timmy. Chapel ... You see: a Book of the Soul and a Prayer Wall ..."
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
Interpretation:
|
||||||
|
- the world exists locally
|
||||||
|
- the first 5 canonical rooms exist
|
||||||
|
- Timmy has a durable home anchor at Gate
|
||||||
|
- a real telnet login/`look` roundtrip works
|
||||||
|
- reconnect lands Timmy back into persisted world state (here seen in Chapel from prior movement)
|
||||||
|
|
||||||
|
## Hermes-in-the-loop proof
|
||||||
|
Local Hermes session:
|
||||||
|
- `20260328_132016_7ea250`
|
||||||
|
|
||||||
|
Prompt intent:
|
||||||
|
- bind session to Evennia telemetry
|
||||||
|
- connect as Timmy
|
||||||
|
- issue world commands through `mcp_evennia_*`
|
||||||
|
|
||||||
|
Observed tool activity included:
|
||||||
|
- `mcp_evennia_bind_session`
|
||||||
|
- `mcp_evennia_connect`
|
||||||
|
- `mcp_evennia_command look`
|
||||||
|
- `mcp_evennia_command enter`
|
||||||
|
- `mcp_evennia_command workshop`
|
||||||
|
- `mcp_evennia_command courtyard`
|
||||||
|
- `mcp_evennia_command chapel`
|
||||||
|
- `mcp_evennia_command look Book of the Soul`
|
||||||
|
|
||||||
|
Important note:
|
||||||
|
- the model’s natural-language summary was imperfect, but the MCP tool calls themselves succeeded and exercised the real world body.
|
||||||
|
- direct bridge proof independently confirmed command results:
|
||||||
|
- `look` at Gate
|
||||||
|
- `enter` to Courtyard
|
||||||
|
- `workshop` to Workshop
|
||||||
|
- `courtyard` back to Courtyard
|
||||||
|
- `chapel` to Chapel
|
||||||
|
- `look Book of the Soul`
|
||||||
|
|
||||||
|
## Visual/operator proof
|
||||||
|
- local browser surface opened at `http://127.0.0.1:4001`
|
||||||
|
- local screenshot captured during verification: `/tmp/evennia-web-proof-2.png`
|
||||||
|
- vision confirmation: screenshot shows the Evennia web home page for `timmy_world`, including the `Play Online` navigation, `Log In` / `Register`, the `Welcome to Evennia!` card, and live server/account stats.
|
||||||
|
|
||||||
|
## Scope delivered in this pass
|
||||||
|
- local Evennia installed and initialized
|
||||||
|
- operator account provisioned
|
||||||
|
- Timmy account/character provisioned
|
||||||
|
- first 5 canonical rooms created
|
||||||
|
- first persistent objects created
|
||||||
|
- Evennia telnet MCP bridge created
|
||||||
|
- launcher script created for local Hermes lane
|
||||||
|
- telemetry helper module added
|
||||||
|
- verification scripts and unit tests added
|
||||||
|
|
||||||
|
## Not fully solved yet
|
||||||
|
- the Hermes prompting around Evennia navigation still needs a tighter skill or system guidance layer so Timmy uses room exits more cleanly and wastes fewer turns
|
||||||
|
- the training lane (#37) is only partially seeded here via telemetry helpers and session binding; replay/eval canon still needs its own pass
|
||||||
14
scripts/evennia/README.md
Normal file
14
scripts/evennia/README.md
Normal file
@@ -0,0 +1,14 @@
|
|||||||
|
# Evennia world lane
|
||||||
|
|
||||||
|
Local runtime target:
|
||||||
|
- `~/.timmy/evennia/timmy_world`
|
||||||
|
|
||||||
|
Main commands:
|
||||||
|
- `python3 scripts/evennia/bootstrap_local_evennia.py`
|
||||||
|
- `python3 scripts/evennia/verify_local_evennia.py`
|
||||||
|
|
||||||
|
Hermes control path:
|
||||||
|
- `scripts/evennia/evennia_mcp_server.py`
|
||||||
|
- intended MCP server key: `evennia`
|
||||||
|
- Timmy acts through a real telnet session to localhost:4000
|
||||||
|
- operator watches through localhost:4001 and/or telnet
|
||||||
140
scripts/evennia/bootstrap_local_evennia.py
Executable file
140
scripts/evennia/bootstrap_local_evennia.py
Executable file
@@ -0,0 +1,140 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||||
|
RUNTIME_ROOT = Path.home() / ".timmy" / "evennia"
|
||||||
|
GAME_DIR = RUNTIME_ROOT / "timmy_world"
|
||||||
|
VENV_DIR = RUNTIME_ROOT / "venv"
|
||||||
|
EVENNIA_BIN = VENV_DIR / "bin" / "evennia"
|
||||||
|
PYTHON_BIN = VENV_DIR / "bin" / "python3"
|
||||||
|
OPERATOR_USER = os.environ.get("TIMMY_EVENNIA_OPERATOR", "Alexander")
|
||||||
|
OPERATOR_EMAIL = os.environ.get("TIMMY_EVENNIA_OPERATOR_EMAIL", "alexpaynex@gmail.com")
|
||||||
|
OPERATOR_PASSWORD = os.environ.get("TIMMY_EVENNIA_OPERATOR_PASSWORD", "timmy-local-dev")
|
||||||
|
TIMMY_PASSWORD = os.environ.get("TIMMY_EVENNIA_TIMMY_PASSWORD", "timmy-world-dev")
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_venv():
|
||||||
|
if not PYTHON_BIN.exists():
|
||||||
|
subprocess.run([sys.executable, "-m", "venv", str(VENV_DIR)], check=True)
|
||||||
|
subprocess.run([str(PYTHON_BIN), "-m", "pip", "install", "--upgrade", "pip"], check=True)
|
||||||
|
subprocess.run([str(PYTHON_BIN), "-m", "pip", "install", "evennia"], check=True)
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_game_dir():
|
||||||
|
if not GAME_DIR.exists():
|
||||||
|
RUNTIME_ROOT.mkdir(parents=True, exist_ok=True)
|
||||||
|
subprocess.run([str(EVENNIA_BIN), "--init", "timmy_world"], cwd=RUNTIME_ROOT, check=True)
|
||||||
|
|
||||||
|
|
||||||
|
def run_evennia(args, env=None, timeout=600):
|
||||||
|
result = subprocess.run([str(EVENNIA_BIN), *args], cwd=GAME_DIR, env=env, timeout=timeout, capture_output=True, text=True)
|
||||||
|
if result.returncode != 0:
|
||||||
|
raise RuntimeError(result.stderr or result.stdout)
|
||||||
|
return result.stdout
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_db_and_owner():
|
||||||
|
run_evennia(["migrate"])
|
||||||
|
env = os.environ.copy()
|
||||||
|
env.update(
|
||||||
|
{
|
||||||
|
"EVENNIA_SUPERUSER_USERNAME": OPERATOR_USER,
|
||||||
|
"EVENNIA_SUPERUSER_EMAIL": OPERATOR_EMAIL,
|
||||||
|
"EVENNIA_SUPERUSER_PASSWORD": OPERATOR_PASSWORD,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
run_evennia(["shell", "-c", "from evennia.accounts.models import AccountDB; print(AccountDB.objects.count())"], env=env)
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_server_started():
|
||||||
|
run_evennia(["start"], timeout=240)
|
||||||
|
|
||||||
|
|
||||||
|
def run_shell(code: str):
|
||||||
|
env = os.environ.copy()
|
||||||
|
env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "")
|
||||||
|
return run_evennia(["shell", "-c", code], env=env)
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_timmy_and_world():
|
||||||
|
code = f'''
|
||||||
|
import sys
|
||||||
|
sys.path.insert(0, {str(REPO_ROOT)!r})
|
||||||
|
from evennia.accounts.models import AccountDB
|
||||||
|
from evennia.accounts.accounts import DefaultAccount
|
||||||
|
from evennia import DefaultRoom, DefaultExit, create_object
|
||||||
|
from evennia.utils.search import search_object
|
||||||
|
from evennia_tools.layout import ROOMS, EXITS, OBJECTS
|
||||||
|
from typeclasses.objects import Object
|
||||||
|
|
||||||
|
acc = AccountDB.objects.filter(username__iexact="Timmy").first()
|
||||||
|
if not acc:
|
||||||
|
acc, errs = DefaultAccount.create(username="Timmy", password={TIMMY_PASSWORD!r})
|
||||||
|
|
||||||
|
room_map = {{}}
|
||||||
|
for room in ROOMS:
|
||||||
|
found = search_object(room.key, exact=True)
|
||||||
|
obj = found[0] if found else None
|
||||||
|
if obj is None:
|
||||||
|
obj, errs = DefaultRoom.create(room.key, description=room.desc)
|
||||||
|
else:
|
||||||
|
obj.db.desc = room.desc
|
||||||
|
room_map[room.key] = obj
|
||||||
|
|
||||||
|
for ex in EXITS:
|
||||||
|
source = room_map[ex.source]
|
||||||
|
dest = room_map[ex.destination]
|
||||||
|
found = [obj for obj in source.contents if obj.key == ex.key and getattr(obj, "destination", None) == dest]
|
||||||
|
if not found:
|
||||||
|
DefaultExit.create(ex.key, source, dest, description=f"Exit to {{dest.key}}.", aliases=list(ex.aliases))
|
||||||
|
|
||||||
|
for spec in OBJECTS:
|
||||||
|
location = room_map[spec.location]
|
||||||
|
found = [obj for obj in location.contents if obj.key == spec.key]
|
||||||
|
if not found:
|
||||||
|
obj = create_object(typeclass=Object, key=spec.key, location=location)
|
||||||
|
else:
|
||||||
|
obj = found[0]
|
||||||
|
obj.db.desc = spec.desc
|
||||||
|
|
||||||
|
char = list(acc.characters)[0]
|
||||||
|
char.location = room_map["Gate"]
|
||||||
|
char.home = room_map["Gate"]
|
||||||
|
char.save()
|
||||||
|
print("WORLD_OK")
|
||||||
|
print("TIMMY_LOCATION", char.location.key)
|
||||||
|
'''
|
||||||
|
return run_shell(code)
|
||||||
|
|
||||||
|
|
||||||
|
def verify():
|
||||||
|
status = run_evennia(["status"], timeout=120)
|
||||||
|
info = run_evennia(["info"], timeout=120)
|
||||||
|
web = subprocess.run(
|
||||||
|
[sys.executable, "-c", "import urllib.request; r=urllib.request.urlopen('http://127.0.0.1:4001', timeout=10); print(r.status); print(r.headers.get('Content-Type'))"],
|
||||||
|
timeout=30,
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
check=True,
|
||||||
|
)
|
||||||
|
return {"status": status.strip(), "info": info.strip(), "web": web.stdout.strip()}
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
ensure_venv()
|
||||||
|
ensure_game_dir()
|
||||||
|
ensure_db_and_owner()
|
||||||
|
ensure_server_started()
|
||||||
|
provision = ensure_timmy_and_world()
|
||||||
|
proof = verify()
|
||||||
|
print(json.dumps({"provision_stdout": provision, "verification": proof}, indent=2))
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
132
scripts/evennia/evennia_mcp_server.py
Executable file
132
scripts/evennia/evennia_mcp_server.py
Executable file
@@ -0,0 +1,132 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import telnetlib
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||||
|
if str(REPO_ROOT) not in sys.path:
|
||||||
|
sys.path.insert(0, str(REPO_ROOT))
|
||||||
|
|
||||||
|
from mcp.server import Server
|
||||||
|
from mcp.server.stdio import stdio_server
|
||||||
|
from mcp.types import Tool, TextContent
|
||||||
|
|
||||||
|
from evennia_tools.telemetry import append_event, excerpt
|
||||||
|
|
||||||
|
HOST = "127.0.0.1"
|
||||||
|
PORT = 4000
|
||||||
|
TIMMY_PASSWORD = os.environ.get("TIMMY_EVENNIA_TIMMY_PASSWORD", "timmy-world-dev")
|
||||||
|
SESSIONS: dict[str, telnetlib.Telnet] = {}
|
||||||
|
STATE_DIR = Path.home() / ".timmy" / "training-data" / "evennia"
|
||||||
|
STATE_FILE = STATE_DIR / "mcp_state.json"
|
||||||
|
app = Server("evennia")
|
||||||
|
|
||||||
|
|
||||||
|
def _load_bound_session_id() -> str:
|
||||||
|
try:
|
||||||
|
if STATE_FILE.exists():
|
||||||
|
data = json.loads(STATE_FILE.read_text(encoding="utf-8"))
|
||||||
|
return (data.get("bound_session_id") or "unbound").strip() or "unbound"
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return "unbound"
|
||||||
|
|
||||||
|
|
||||||
|
def _save_bound_session_id(session_id: str) -> str:
|
||||||
|
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
||||||
|
session_id = (session_id or "unbound").strip() or "unbound"
|
||||||
|
STATE_FILE.write_text(json.dumps({"bound_session_id": session_id}, indent=2), encoding="utf-8")
|
||||||
|
return session_id
|
||||||
|
|
||||||
|
|
||||||
|
def _read(tn: telnetlib.Telnet, delay: float = 0.5) -> str:
|
||||||
|
time.sleep(delay)
|
||||||
|
return tn.read_very_eager().decode("utf-8", "ignore")
|
||||||
|
|
||||||
|
|
||||||
|
def _connect(name: str = "timmy", username: str = "Timmy", password: str | None = None) -> dict:
|
||||||
|
if name in SESSIONS:
|
||||||
|
return {"connected": True, "name": name, "output": ""}
|
||||||
|
tn = telnetlib.Telnet(HOST, PORT, timeout=10)
|
||||||
|
banner = _read(tn, 1.0)
|
||||||
|
tn.write(f"connect {username} {password or TIMMY_PASSWORD}\n".encode())
|
||||||
|
out = _read(tn, 1.0)
|
||||||
|
SESSIONS[name] = tn
|
||||||
|
append_event(_load_bound_session_id(), {"event": "connect", "actor": username, "output": excerpt(banner + "\n" + out)})
|
||||||
|
return {"connected": True, "name": name, "output": banner + "\n" + out}
|
||||||
|
|
||||||
|
|
||||||
|
def _observe(name: str = "timmy") -> dict:
|
||||||
|
if name not in SESSIONS:
|
||||||
|
_connect(name=name)
|
||||||
|
tn = SESSIONS[name]
|
||||||
|
return {"name": name, "output": _read(tn, 0.2)}
|
||||||
|
|
||||||
|
|
||||||
|
def _command(command: str, name: str = "timmy", wait_ms: int = 500) -> dict:
|
||||||
|
if name not in SESSIONS:
|
||||||
|
_connect(name=name)
|
||||||
|
tn = SESSIONS[name]
|
||||||
|
tn.write((command + "\n").encode())
|
||||||
|
out = _read(tn, max(0.1, wait_ms / 1000.0))
|
||||||
|
append_event(_load_bound_session_id(), {"event": "command", "actor": name, "command": command, "output": excerpt(out, 1000)})
|
||||||
|
return {"name": name, "command": command, "output": out}
|
||||||
|
|
||||||
|
|
||||||
|
def _disconnect(name: str = "timmy") -> dict:
|
||||||
|
tn = SESSIONS.pop(name, None)
|
||||||
|
if tn:
|
||||||
|
try:
|
||||||
|
tn.close()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
append_event(_load_bound_session_id(), {"event": "disconnect", "actor": name})
|
||||||
|
return {"disconnected": bool(tn), "name": name}
|
||||||
|
|
||||||
|
|
||||||
|
@app.list_tools()
|
||||||
|
async def list_tools():
|
||||||
|
return [
|
||||||
|
Tool(name="bind_session", description="Bind a Hermes session id to Evennia telemetry logs.", inputSchema={"type": "object", "properties": {"session_id": {"type": "string"}}, "required": ["session_id"]}),
|
||||||
|
Tool(name="status", description="Show Evennia MCP/telnet control status.", inputSchema={"type": "object", "properties": {}, "required": []}),
|
||||||
|
Tool(name="connect", description="Connect Timmy to the local Evennia telnet server as a real in-world account.", inputSchema={"type": "object", "properties": {"name": {"type": "string"}, "username": {"type": "string"}, "password": {"type": "string"}}, "required": []}),
|
||||||
|
Tool(name="observe", description="Read pending text output from Timmy's Evennia connection.", inputSchema={"type": "object", "properties": {"name": {"type": "string"}}, "required": []}),
|
||||||
|
Tool(name="command", description="Send one Evennia command line as Timmy and return the resulting text output.", inputSchema={"type": "object", "properties": {"command": {"type": "string"}, "name": {"type": "string"}, "wait_ms": {"type": "integer", "default": 500}}, "required": ["command"]}),
|
||||||
|
Tool(name="disconnect", description="Close Timmy's Evennia telnet control session.", inputSchema={"type": "object", "properties": {"name": {"type": "string"}}, "required": []}),
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@app.call_tool()
|
||||||
|
async def call_tool(name: str, arguments: dict):
|
||||||
|
arguments = arguments or {}
|
||||||
|
if name == "bind_session":
|
||||||
|
bound = _save_bound_session_id(arguments.get("session_id", "unbound"))
|
||||||
|
result = {"bound_session_id": bound}
|
||||||
|
elif name == "status":
|
||||||
|
result = {"connected_sessions": sorted(SESSIONS.keys()), "bound_session_id": _load_bound_session_id()}
|
||||||
|
elif name == "connect":
|
||||||
|
result = _connect(arguments.get("name", "timmy"), arguments.get("username", "Timmy"), arguments.get("password"))
|
||||||
|
elif name == "observe":
|
||||||
|
result = _observe(arguments.get("name", "timmy"))
|
||||||
|
elif name == "command":
|
||||||
|
result = _command(arguments.get("command", ""), arguments.get("name", "timmy"), arguments.get("wait_ms", 500))
|
||||||
|
elif name == "disconnect":
|
||||||
|
result = _disconnect(arguments.get("name", "timmy"))
|
||||||
|
else:
|
||||||
|
result = {"error": f"Unknown tool: {name}"}
|
||||||
|
return [TextContent(type="text", text=json.dumps(result, indent=2))]
|
||||||
|
|
||||||
|
|
||||||
|
async def main():
|
||||||
|
async with stdio_server() as (read_stream, write_stream):
|
||||||
|
await app.run(read_stream, write_stream, app.create_initialization_options())
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
import asyncio
|
||||||
|
asyncio.run(main())
|
||||||
70
scripts/evennia/launch_timmy_evennia_local.sh
Executable file
70
scripts/evennia/launch_timmy_evennia_local.sh
Executable file
@@ -0,0 +1,70 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
export HERMES_HOME="${HERMES_HOME:-$HOME/.hermes-local}"
|
||||||
|
SRC_CFG="$HOME/.hermes/config.yaml"
|
||||||
|
DST_CFG="$HERMES_HOME/config.yaml"
|
||||||
|
REPO_ROOT="${REPO_ROOT:-$HOME/.timmy}"
|
||||||
|
|
||||||
|
mkdir -p "$HERMES_HOME"
|
||||||
|
if [ ! -f "$DST_CFG" ]; then
|
||||||
|
cp "$SRC_CFG" "$DST_CFG"
|
||||||
|
fi
|
||||||
|
if [ ! -e "$HERMES_HOME/skins" ]; then
|
||||||
|
ln -s "$HOME/.hermes/skins" "$HERMES_HOME/skins"
|
||||||
|
fi
|
||||||
|
|
||||||
|
python3 - <<'PY'
|
||||||
|
from pathlib import Path
|
||||||
|
import yaml
|
||||||
|
|
||||||
|
home = Path.home()
|
||||||
|
cfg_path = home / '.hermes-local' / 'config.yaml'
|
||||||
|
data = yaml.safe_load(cfg_path.read_text()) or {}
|
||||||
|
|
||||||
|
model = data.get('model')
|
||||||
|
if not isinstance(model, dict):
|
||||||
|
model = {'default': str(model)} if model else {}
|
||||||
|
data['model'] = model
|
||||||
|
model['default'] = 'NousResearch_Hermes-4-14B-Q4_K_M.gguf'
|
||||||
|
model['provider'] = 'custom'
|
||||||
|
model['base_url'] = 'http://localhost:8081/v1'
|
||||||
|
model['context_length'] = 65536
|
||||||
|
|
||||||
|
providers = data.get('custom_providers')
|
||||||
|
if not isinstance(providers, list):
|
||||||
|
providers = []
|
||||||
|
data['custom_providers'] = providers
|
||||||
|
found = False
|
||||||
|
for entry in providers:
|
||||||
|
if isinstance(entry, dict) and entry.get('name') == 'Local llama.cpp':
|
||||||
|
entry['base_url'] = 'http://localhost:8081/v1'
|
||||||
|
entry['api_key'] = 'none'
|
||||||
|
entry['model'] = 'NousResearch_Hermes-4-14B-Q4_K_M.gguf'
|
||||||
|
found = True
|
||||||
|
break
|
||||||
|
if not found:
|
||||||
|
providers.insert(0, {
|
||||||
|
'name': 'Local llama.cpp',
|
||||||
|
'base_url': 'http://localhost:8081/v1',
|
||||||
|
'api_key': 'none',
|
||||||
|
'model': 'NousResearch_Hermes-4-14B-Q4_K_M.gguf',
|
||||||
|
})
|
||||||
|
|
||||||
|
import os
|
||||||
|
repo_root = Path(os.environ.get('REPO_ROOT', str(Path.home() / '.timmy'))).expanduser()
|
||||||
|
script_path = repo_root / 'scripts' / 'evennia' / 'evennia_mcp_server.py'
|
||||||
|
|
||||||
|
data['mcp_servers'] = {
|
||||||
|
'evennia': {
|
||||||
|
'command': 'python3',
|
||||||
|
'args': [str(script_path)],
|
||||||
|
'env': {},
|
||||||
|
'timeout': 30,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
cfg_path.write_text(yaml.safe_dump(data, sort_keys=False, allow_unicode=True))
|
||||||
|
PY
|
||||||
|
|
||||||
|
exec hermes chat "$@"
|
||||||
66
scripts/evennia/verify_local_evennia.py
Executable file
66
scripts/evennia/verify_local_evennia.py
Executable file
@@ -0,0 +1,66 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
|
import telnetlib
|
||||||
|
import time
|
||||||
|
import urllib.request
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parents[2]
|
||||||
|
ROOT = Path.home() / ".timmy" / "evennia" / "timmy_world"
|
||||||
|
EVENNIA_BIN = Path.home() / ".timmy" / "evennia" / "venv" / "bin" / "evennia"
|
||||||
|
TIMMY_PASSWORD = os.environ.get("TIMMY_EVENNIA_TIMMY_PASSWORD", "timmy-world-dev")
|
||||||
|
|
||||||
|
|
||||||
|
def shell_json(code: str) -> dict:
|
||||||
|
env = dict(**os.environ)
|
||||||
|
env["PYTHONPATH"] = str(REPO_ROOT) + ":" + env.get("PYTHONPATH", "")
|
||||||
|
result = subprocess.run([str(EVENNIA_BIN), "shell", "-c", code], cwd=ROOT, env=env, capture_output=True, text=True, timeout=120, check=True)
|
||||||
|
lines = [line for line in result.stdout.splitlines() if line.strip()]
|
||||||
|
return json.loads(lines[-1])
|
||||||
|
|
||||||
|
|
||||||
|
def telnet_roundtrip() -> str:
|
||||||
|
tn = telnetlib.Telnet("127.0.0.1", 4000, timeout=10)
|
||||||
|
time.sleep(1.0)
|
||||||
|
_ = tn.read_very_eager().decode("utf-8", "ignore")
|
||||||
|
tn.write(f"connect Timmy {TIMMY_PASSWORD}\n".encode())
|
||||||
|
time.sleep(1.0)
|
||||||
|
login_out = tn.read_very_eager().decode("utf-8", "ignore")
|
||||||
|
tn.write(b"look\n")
|
||||||
|
time.sleep(0.6)
|
||||||
|
look_out = tn.read_very_eager().decode("utf-8", "ignore")
|
||||||
|
tn.close()
|
||||||
|
return " ".join((login_out + "\n" + look_out).split())[:600]
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
result = {"game_dir_exists": ROOT.exists()}
|
||||||
|
try:
|
||||||
|
with urllib.request.urlopen("http://127.0.0.1:4001", timeout=10) as r:
|
||||||
|
result["web_status"] = r.status
|
||||||
|
result["web_content_type"] = r.headers.get("Content-Type")
|
||||||
|
except Exception as e:
|
||||||
|
result["web_error"] = f"{type(e).__name__}: {e}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
state = shell_json(
|
||||||
|
"from evennia.accounts.models import AccountDB; from evennia.utils.search import search_object; "
|
||||||
|
"acc=AccountDB.objects.get(username='Timmy'); char=list(acc.characters)[0]; rooms=['Gate','Courtyard','Workshop','Archive','Chapel']; "
|
||||||
|
"import json; print(json.dumps({'timmy_home': getattr(char.home,'key',None), 'rooms': {name: bool(search_object(name, exact=True)) for name in rooms}}))"
|
||||||
|
)
|
||||||
|
result.update(state)
|
||||||
|
except Exception as e:
|
||||||
|
result["world_error"] = f"{type(e).__name__}: {e}"
|
||||||
|
|
||||||
|
try:
|
||||||
|
result["telnet_roundtrip_excerpt"] = telnet_roundtrip()
|
||||||
|
except Exception as e:
|
||||||
|
result["telnet_error"] = f"{type(e).__name__}: {e}"
|
||||||
|
|
||||||
|
print(json.dumps(result, indent=2))
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
28
tests/test_evennia_layout.py
Normal file
28
tests/test_evennia_layout.py
Normal file
@@ -0,0 +1,28 @@
|
|||||||
|
import unittest
|
||||||
|
|
||||||
|
from evennia_tools.layout import EXITS, OBJECTS, grouped_exits, room_keys
|
||||||
|
|
||||||
|
|
||||||
|
class TestEvenniaLayout(unittest.TestCase):
|
||||||
|
def test_first_wave_rooms_are_exact(self):
|
||||||
|
self.assertEqual(room_keys(), ("Gate", "Courtyard", "Workshop", "Archive", "Chapel"))
|
||||||
|
|
||||||
|
def test_all_exit_endpoints_exist(self):
|
||||||
|
keys = set(room_keys())
|
||||||
|
for ex in EXITS:
|
||||||
|
self.assertIn(ex.source, keys)
|
||||||
|
self.assertIn(ex.destination, keys)
|
||||||
|
|
||||||
|
def test_courtyard_is_navigation_hub(self):
|
||||||
|
exits = grouped_exits()["Courtyard"]
|
||||||
|
destinations = {ex.destination for ex in exits}
|
||||||
|
self.assertEqual(destinations, {"Gate", "Workshop", "Archive", "Chapel"})
|
||||||
|
|
||||||
|
def test_objects_live_in_known_rooms(self):
|
||||||
|
keys = set(room_keys())
|
||||||
|
for obj in OBJECTS:
|
||||||
|
self.assertIn(obj.location, keys)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
28
tests/test_evennia_telemetry.py
Normal file
28
tests/test_evennia_telemetry.py
Normal file
@@ -0,0 +1,28 @@
|
|||||||
|
import json
|
||||||
|
import tempfile
|
||||||
|
import unittest
|
||||||
|
|
||||||
|
from evennia_tools.telemetry import append_event, event_log_path, excerpt
|
||||||
|
|
||||||
|
|
||||||
|
class TestEvenniaTelemetry(unittest.TestCase):
|
||||||
|
def test_event_log_path_contains_session_id(self):
|
||||||
|
path = event_log_path("session_abc", "/tmp/evennia-test")
|
||||||
|
self.assertEqual(path.name, "session_abc.jsonl")
|
||||||
|
|
||||||
|
def test_append_event_writes_jsonl(self):
|
||||||
|
with tempfile.TemporaryDirectory() as td:
|
||||||
|
path = append_event("session_xyz", {"event": "look", "room": "Gate"}, td)
|
||||||
|
self.assertTrue(path.exists())
|
||||||
|
line = path.read_text(encoding="utf-8").strip()
|
||||||
|
data = json.loads(line)
|
||||||
|
self.assertEqual(data["event"], "look")
|
||||||
|
self.assertEqual(data["room"], "Gate")
|
||||||
|
self.assertIn("timestamp", data)
|
||||||
|
|
||||||
|
def test_excerpt_compacts_whitespace(self):
|
||||||
|
self.assertEqual(excerpt("a b\n c"), "a b c")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user