lycis/mistle
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: call agent execution from interactive mode

Browse source
This commit is contained in:
Daniel Eder 2025-09-27 08:21:10 +02:00
parent 9d87390246
commit ed4e697ad0
3 changed files with 70 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.env
View file

@ -4,5 +4,5 @@ MISTLE_USER=mistle
MISTLE_PASSWORD=sl-mudbot
MISTLE_LOGIN_PROMPT=Wie heisst Du denn ("neu" fuer neuen Spieler) ?
MISTLE_EXIT_COMMAND=schlaf ein
MISTLE_AGENT_MODE=true
MISTLE_AGENT_MODE=false
MISTLE_AGENT=explore

25
README.md
View file

@ -1,14 +1,14 @@
# Mistle Mudbot
Python-based Telnet helper for connecting to MUD servers, handling login flows, and optionally running an automated "agent" alongside interactive play.
Python-based Telnet helper for connecting to MUD servers, handling login flows, and optionally running automated "agents" alongside interactive play.
## Features
- Lightweight wrapper (`TelnetClient`) around `telnetlib` with sane defaults and context-manager support.
- Loads credentials and connection settings from a local `.env` file.
- Interactive console session that mirrors server output and lets you type commands directly.
- Optional agent mode that runs on a background thread and can dispatch automated commands without blocking manual input.
- Pluggable agent selection via environment variables (built-in `SimpleAgent`, extendable with custom classes).
- Optional always-on agent mode plus an on-demand `#execute <agent>` escape hatch for ad-hoc automations.
- Built-in agents (`SimpleAgent`, `ExploreAgent`) with a pluggable interface for custom behaviours.
## Requirements
@ -33,6 +33,14 @@ Python-based Telnet helper for connecting to MUD servers, handling login flows,
4. Type commands directly into the console. Press `Ctrl-C` to exit; the client will send any configured shutdown command to the MUD.
5. When *not* running in agent mode you can kick off one-off automations from the prompt:
```text
#execute explore
```
The command remains interactive while the agent works in the background and stops automatically a few seconds after things quiet down.
## Environment Variables
All variables can be placed in the `.env` file (one `KEY=value` per line) or provided through the shell environment.
@ -45,17 +53,24 @@ All variables can be placed in the `.env` file (one `KEY=value` per line) or pro
| `MISTLE_PASSWORD` | ❌ | Password sent after the username. Leave blank for manual entry. |
| `MISTLE_LOGIN_PROMPT` | ❌ | Prompt string that signals the client to send credentials (e.g., `"Name:"`). When omitted, the client just waits for the initial banner. |
| `MISTLE_EXIT_COMMAND` | ❌ | Command issued during graceful shutdown (after pressing `Ctrl-C`). Useful for `quit`/`save` macros. |
| `MISTLE_AGENT_MODE` | ❌ | Enable agent thread when set to truthy values (`1`, `true`, `yes`, `on`). Defaults to interactive-only mode. |
| `MISTLE_AGENT_MODE` | ❌ | Enable full-time agent thread when set to truthy values (`1`, `true`, `yes`, `on`). Defaults to interactive-only mode. |
| `MISTLE_AGENT` | ❌ | Select which agent class to instantiate when agent mode is active. Accepted values: `simple` (default), `explore` (requires `ExploreAgent` inside `agent.py`), or custom spec `module:ClassName`. |
## Agent Development
- Implement new agents by subclassing `agent.Agent` and overriding `observe()` and `decide()`.
- Register the agent by either:
- Adding the class to `agent.py` and referencing it in `MISTLE_AGENT` (e.g., `explore` if the class is named `ExploreAgent`).
- Adding the class to `agent.py` and referencing it in `MISTLE_AGENT` (e.g., `explore` for `ExploreAgent`).
- Placing the class elsewhere and configuring `MISTLE_AGENT` to `your_module:YourAgent`.
- `observe(output)` receives the latest server text; `decide()` returns the next command string or `None` to stay idle.
- Commands issued by the agent are throttled to one per second so manual commands can still interleave smoothly.
- `ExploreAgent` showcases a richer workflow: it sends `schau`, identifies German nouns, inspects each with `untersuche`, and prints `[Agent]` progress updates like `Explored 3/7 — untersuche Tisch`.
## On-Demand Agents
- When `MISTLE_AGENT_MODE` is **off**, you can trigger an ephemeral agent at any time with `#execute <agent_spec>`.
- The syntax accepts the same values as `MISTLE_AGENT` and reuses the `build_agent` helper, so `#execute simple`, `#execute explore`, or `#execute mypackage.mymodule:CustomAgent` are all valid.
- On-demand runs share the current session, respect the one-command-per-second limit, and stop automatically after a few seconds of inactivity.
## Danger Zone

50
app.py
View file

@ -5,7 +5,7 @@ import time
from importlib import import_module
from pathlib import Path
from threading import Event, Lock, Thread
from typing import Optional, Type
from typing import Callable, Optional, Type
from agent import Agent, SimpleAgent
from telnetclient import TelnetClient
@ -75,14 +75,26 @@ def run_agent_loop(
*,
idle_delay: float = 0.5,
min_send_interval: float = 1.0,
auto_stop: bool = False,
auto_stop_idle: float = 2.0,
) -> None:
"""Invoke *agent* whenever new output arrives and send its response."""
idle_started: Optional[float] = None
while not stop_event.is_set():
triggered = state.wait_for_output(timeout=idle_delay)
if stop_event.is_set():
break
if not triggered:
if auto_stop:
now = time.time()
if idle_started is None:
idle_started = now
elif now - idle_started >= auto_stop_idle:
break
continue
idle_started = None
state.clear_output_event()
last_output = state.snapshot_output()
if not last_output:
@ -209,6 +221,7 @@ def interactive_session(
poll_interval: float = 0.2,
receive_timeout: float = 0.2,
exit_command: str,
agent_command: Optional[Callable[[str], None]] = None,
) -> None:
"""Keep the Telnet session running, proxying input/output until interrupted."""
if exit_command:
@ -231,6 +244,13 @@ def interactive_session(
line = line.rstrip("\r\n")
if not line:
continue
if agent_command and line.lower().startswith("#execute"):
parts = line.split(maxsplit=1)
if len(parts) == 1:
print("[Agent] Usage: #execute <agent_spec>")
else:
agent_command(parts[1])
continue
state.send(client, line)
@ -278,6 +298,7 @@ def main() -> int:
stop_event = Event()
agent_thread: Optional[Thread] = None
agent: Optional[Agent] = None
ephemeral_agents: list[Thread] = []
with TelnetClient(host=host, port=port, timeout=10.0) as client:
login(
@ -298,6 +319,30 @@ def main() -> int:
)
agent_thread.start()
def run_ephemeral_agent(spec: str) -> None:
spec = spec.strip()
if not spec:
print("[Agent] Usage: #execute <agent_spec>")
return
try:
temp_agent = build_agent(spec)
except RuntimeError as exc:
print(f"[Agent] Failed to load '{spec}': {exc}", file=sys.stderr)
return
thread = Thread(
target=run_agent_loop,
args=(client, state, temp_agent, stop_event),
kwargs={
"min_send_interval": 1.0,
"auto_stop": True,
"auto_stop_idle": 2.0,
},
daemon=True,
)
ephemeral_agents.append(thread)
print(f"[Agent] Executing {spec!r} once")
thread.start()
interrupted = False
try:
interactive_session(
@ -305,6 +350,7 @@ def main() -> int:
state=state,
stop_event=stop_event,
exit_command=exit_command,
agent_command=None if agent_mode else run_ephemeral_agent,
)
except KeyboardInterrupt:
print()
@ -313,6 +359,8 @@ def main() -> int:
stop_event.set()
if agent_thread:
agent_thread.join(timeout=1.0)
for thread in ephemeral_agents:
thread.join(timeout=1.0)
if interrupted:
graceful_shutdown(client, exit_command, state=state)