chore: removed .env

.env

@@ -1,7 +0,0 @@
-MISTLE_HOST=sl.mud.at
-MISTLE_PORT=4711
-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

.gitignore

@@ -9,3 +9,6 @@ wheels/
 # Virtual environments
 .venv
 .DS_STORE
+
+# config
+.env

.vscode/settings.json

@@ -0,0 +1,3 @@
+{
+    "wolf.disableHotModeWarning": true
+}

README.md

@@ -0,0 +1,89 @@
+# Mistle Mudbot
+
+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 always-on agent mode plus an on-demand `#execute <agent>` escape hatch for ad-hoc automations.
+- Built-in agents (`SimpleAgent`, `ExploreAgent`, `CommunicationAgent`, `IntelligentCommunicationAgent`) with a pluggable interface for custom behaviours.
+
+## Requirements
+
+- Python 3.10+
+- A reachable Telnet-based MUD server
+
+## Quick Start
+
+1. Create a virtual environment and install dependencies:
+
+   ```bash
+   pip install -e .
+   ```
+
+2. Copy `.env.example` (if available) or create a `.env` file in the project root, then fill in the connection details described below.
+
+3. Run the client:
+
+   ```bash
+   python app.py
+   ```
+
+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.
+
+| Variable | Required | Description |
+| --- | --- | --- |
+| `MISTLE_HOST` | ✅ | DNS name or IP of the target MUD server. |
+| `MISTLE_PORT` | ✅ | Telnet port number (will be cast to `int`). |
+| `MISTLE_USER` | ❌ | Username or character name; sent automatically after login banner, if provided. |
+| `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 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`, `communication`, `intelligent`/`intelligentcommunication` (LLM-backed), or custom spec `module:ClassName`. |
+| `MISTLE_LLM_MODEL` | ❌ | Override the `litellm` model used by the intelligent agent (defaults to `mistral/mistral-small-2407`). |
+| `MISTRAL_API_KEY` | ❌ | API key used by `IntelligentCommunicationAgent` (via `litellm`) when calling the `mistral/mistral-small-2407` model. |
+
+## 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` 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`.
+- `CommunicationAgent` auto-replies to every direct tell with a canned greeting, while `IntelligentCommunicationAgent` routes each tell through `litellm` (default model `mistral/mistral-small-2407`) to craft a contextual answer.
+
+## 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`, `#execute intelligent`, 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
+
+- The Telnet session runs until you interrupt it. Make sure the terminal is in a state where `Ctrl-C` is available.
+- When adding new agents, guard any long-running logic to avoid blocking the agent thread.
+
+## Contributing
+
+Feel free to open issues or submit pull requests for additional MUD-specific helpers, new agents, or quality-of-life improvements.
+
+---
+
+Happy MUDding!

agent.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import re
 from abc import ABC, abstractmethod
+from collections import deque
 from dataclasses import dataclass, field
-from typing import Optional
+from typing import Deque, Optional, Pattern, Set, Tuple
 
 
 class Agent(ABC):
@@ -19,7 +21,7 @@ class Agent(ABC):
 
 @dataclass
 class SimpleAgent(Agent):
-    """Very small stateful agent that decides which command to run next."""
+    """Minimal agent that always returns the same command."""
 
     default_command: str = "schau"
     last_output: str = field(default="", init=False)
@@ -30,3 +32,86 @@ class SimpleAgent(Agent):
 
     def decide(self) -> Optional[str]:
         return self.default_command
+
+
+@dataclass
+class ExploreAgent(Agent):
+    """Agent that inspects every noun it discovers in the room description."""
+
+    look_command: str = "schau"
+    inspect_command: str = "untersuche"
+    nouns_pattern: Pattern[str] = field(
+        default_factory=lambda: re.compile(r"\b[A-ZÄÖÜ][A-Za-zÄÖÜäöüß-]*\b")
+    )
+    last_output: str = field(default="", init=False)
+    look_sent: bool = field(default=False, init=False)
+    pending_targets: Deque[str] = field(default_factory=deque, init=False)
+    seen_targets: Set[str] = field(default_factory=set, init=False)
+    inspected_targets: Set[str] = field(default_factory=set, init=False)
+
+    def observe(self, output: str) -> None:
+        if not output:
+            return
+        self.last_output = output
+        if not self.look_sent:
+            return
+
+        for noun in self.nouns_pattern.findall(output):
+            target = noun.strip()
+            key = target.lower()
+            if not target or key in self.seen_targets:
+                continue
+            self.seen_targets.add(key)
+            self.pending_targets.append(target)
+
+    def decide(self) -> Optional[str]:
+        if not self.look_sent:
+            self.look_sent = True
+            print("[Agent] Exploring room, sending 'schau'")
+            return self.look_command
+
+        if self.pending_targets:
+            target = self.pending_targets.popleft()
+            key = target.lower()
+            self.inspected_targets.add(key)
+            progress = f"[Agent] Explored {len(self.inspected_targets)}/{len(self.seen_targets)} — untersuche {target}"
+            print(progress)
+            return f"{self.inspect_command} {target}"
+
+        return None
+
+
+@dataclass
+class CommunicationAgent(Agent):
+    """Agent that replies to private tells."""
+
+    reply_template: str = "teile {player} mit Hallo! Ich bin Mistle und ein Bot."
+    tell_pattern: Pattern[str] = field(
+        default_factory=lambda: re.compile(
+            r"^(?P<player>[^\s]+) teilt (d|D)ir mit: (?P<message>.+)$",
+            re.MULTILINE,
+        )
+    )
+    last_output: str = field(default="", init=False)
+    pending_replies: Deque[Tuple[str, str]] = field(default_factory=deque, init=False)
+
+    def observe(self, output: str) -> None:
+        if not output:
+            return
+        self.last_output = output
+        for match in self.tell_pattern.finditer(output):
+            player = match.group("player").strip()
+            message = match.group("message").strip()
+            if not player:
+                continue
+            self.pending_replies.append((player, message))
+            print(f"[Agent] Received message from {player}: {message}")
+
+    def decide(self) -> Optional[str]:
+        if not self.pending_replies:
+            return None
+        player, _ = self.pending_replies.popleft()
+        reply = self.reply_template.format(player=player)
+        print(f"[Agent] Replying to {player}")
+        return reply
+

app.py

@@ -2,9 +2,10 @@ import os
 import select
 import sys
 import time
+from importlib import import_module
 from pathlib import Path
 from threading import Event, Lock, Thread
-from typing import Optional
+from typing import Callable, Optional, Type
 
 from agent import Agent, SimpleAgent
 from telnetclient import TelnetClient
@@ -74,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:
@@ -129,6 +142,67 @@ def require_env(key: str) -> str:
     return value
 
 
+def build_agent(agent_spec: str) -> Agent:
+    """Instantiate an agent based on ``MISTLE_AGENT`` contents."""
+    normalized = agent_spec.strip()
+    if not normalized:
+        return SimpleAgent()
+
+    key = normalized.lower()
+    if key == "simple":
+        return SimpleAgent()
+
+    builtin_agents = {
+        "explore": ("agent", "ExploreAgent", {}),
+        "communication": ("agent", "CommunicationAgent", {}),
+        "intelligent": (
+            "intelligent_agent",
+            "IntelligentCommunicationAgent",
+            {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-tiny")},
+        ),
+        "intelligentcommunication": (
+            "intelligent_agent",
+            "IntelligentCommunicationAgent",
+            {"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-tiny")},
+        ),
+    }
+
+    if key in builtin_agents:
+        module_name, class_name, kwargs = builtin_agents[key]
+        try:
+            module = import_module(module_name)
+            agent_cls = getattr(module, class_name)
+        except AttributeError as exc:  # pragma: no cover - optional dependency
+            raise RuntimeError(f"{class_name} is not available in agent module") from exc
+        agent = _instantiate_agent(agent_cls, normalized, kwargs)
+        model_name = kwargs.get("model") if kwargs else None
+        if model_name:
+            print(f"[Agent] Using LLM model: {model_name}")
+        return agent
+
+    if ":" in normalized:
+        module_name, class_name = normalized.split(":", 1)
+        if not module_name or not class_name:
+            raise RuntimeError("MISTLE_AGENT must be in 'module:ClassName' format")
+        module = import_module(module_name)
+        agent_cls = getattr(module, class_name)
+        return _instantiate_agent(agent_cls, normalized)
+
+    raise RuntimeError(f"Unknown agent spec '{agent_spec}'.")
+
+
+def _instantiate_agent(
+    agent_cls: Type[Agent], agent_spec: str, kwargs: Optional[dict] = None
+) -> Agent:
+    if not issubclass(agent_cls, Agent):
+        raise RuntimeError(f"{agent_spec} is not an Agent subclass")
+    try:
+        kwargs = kwargs or {}
+        return agent_cls(**kwargs)
+    except TypeError as exc:
+        raise RuntimeError(f"Failed to instantiate {agent_spec}: {exc}") from exc
+
+
 def login(
     client: TelnetClient,
     *,
@@ -170,6 +244,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:
@@ -192,6 +267,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)
 
 
@@ -233,11 +315,13 @@ def main() -> int:
     login_prompt = os.environ.get("MISTLE_LOGIN_PROMPT", "")
     exit_command = os.environ.get("MISTLE_EXIT_COMMAND", "")
     agent_mode = os.environ.get("MISTLE_AGENT_MODE", "").lower() in {"1", "true", "yes", "on"}
+    agent_spec = os.environ.get("MISTLE_AGENT", "")
 
     state = SessionState()
     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(
@@ -249,7 +333,7 @@ def main() -> int:
         )
 
         if agent_mode:
-            agent = SimpleAgent()
+            agent = build_agent(agent_spec)
             agent_thread = Thread(
                 target=run_agent_loop,
                 args=(client, state, agent, stop_event),
@@ -258,6 +342,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(
@@ -265,6 +373,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()
@@ -273,6 +382,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)

intelligent_agent.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import re
+import sys
+from collections import deque
+from dataclasses import dataclass, field
+from typing import Deque, Dict, Optional, Pattern, Tuple
+
+try:
+    from litellm import completion
+except ImportError:  # pragma: no cover - optional dependency
+    completion = None  # type: ignore[assignment]
+
+from agent import Agent
+
+
+@dataclass
+class IntelligentCommunicationAgent(Agent):
+    """Agent that uses a language model to answer private tells."""
+
+    model: str = "mistral/mistral-tiny"
+    system_prompt: str = (
+        "Du bist Mistle, ein hilfsbereiter MUD-Bot. "
+        "Antworte freundlich und knapp in deutscher Sprache."
+    )
+    temperature: float = 0.7
+    max_output_tokens: int = 120
+    fallback_reply: str = "Hallo! Ich bin Mistle und ein Bot."
+    tell_pattern: Pattern[str] = field(
+        default_factory=lambda: re.compile(
+            r"^(?P<player>[^\s]+) teilt (d|D)ir mit: (?P<message>.+)$",
+            re.MULTILINE,
+        )
+    )
+    last_output: str = field(default="", init=False)
+    max_history_chars: int = 16_000
+    pending_replies: Deque[Tuple[str, str]] = field(default_factory=deque, init=False)
+    conversation_history: Dict[str, Deque[Tuple[str, str]]] = field(default_factory=dict, init=False)
+
+    def observe(self, output: str) -> None:
+        if not output:
+            return
+        self.last_output = output
+        for match in self.tell_pattern.finditer(output):
+            player = match.group("player").strip()
+            message = match.group("message").strip()
+            if not player:
+                continue
+            self.pending_replies.append((player, message))
+            print(f"[Agent] Received message from {player}: {message}")
+            self._append_history(player, "user", message)
+
+    def decide(self) -> Optional[str]:
+        if not self.pending_replies:
+            return None
+        player, _ = self.pending_replies.popleft()
+        reply_text = self._sanitize_reply(self._generate_reply(player))
+        self._append_history(player, "assistant", reply_text)
+        reply = f"teile {player} mit {reply_text}"
+        print(f"[Agent] Replying to {player} with model output")
+        return reply
+
+    def _generate_reply(self, player: str) -> str:
+        if completion is None:
+            print(
+                "[Agent] litellm is not installed; falling back to default reply",
+                file=sys.stderr,
+            )
+            return self.fallback_reply
+
+        try:
+            response = completion(
+                model=self.model,
+                messages=self._build_messages(player),
+                temperature=self.temperature,
+                max_tokens=self.max_output_tokens,
+            )
+        except Exception as exc:  # pragma: no cover - network/runtime errors
+            print(f"[Agent] Model call failed: {exc}", file=sys.stderr)
+            return self.fallback_reply
+
+        try:
+            content = response["choices"][0]["message"]["content"].strip()
+        except (KeyError, IndexError, TypeError):  # pragma: no cover - defensive
+            return self.fallback_reply
+
+        return content or self.fallback_reply
+
+    def _sanitize_reply(self, text: str) -> str:
+        if not text:
+            return self.fallback_reply
+        collapsed = " ".join(text.split())
+        return collapsed or self.fallback_reply
+
+    def _build_messages(self, player: str) -> list[dict[str, str]]:
+        history = self.conversation_history.get(player)
+        messages: list[dict[str, str]] = [{"role": "system", "content": self.system_prompt}]
+        if not history:
+            return messages
+        for role, content in history:
+            messages.append({"role": role, "content": content})
+        return messages
+
+    def _append_history(self, player: str, role: str, content: str) -> None:
+        if not content:
+            return
+        history = self.conversation_history.setdefault(player, deque())
+        history.append((role, content))
+        self._trim_history(history)
+
+    def _trim_history(self, history: Deque[Tuple[str, str]]) -> None:
+        total_chars = sum(len(content) for _, content in history)
+        while len(history) > 1 and total_chars > self.max_history_chars:
+            _, removed_content = history.popleft()
+            total_chars -= len(removed_content)
+        if history and total_chars > self.max_history_chars:
+            role, content = history.pop()
+            trimmed = content[-self.max_history_chars :]
+            history.append((role, trimmed))

pyproject.toml

@@ -4,4 +4,6 @@ version = "0.1.0"
 description = "Add your description here"
 readme = "README.md"
 requires-python = ">=3.9"
-dependencies = []
+dependencies = [
+    "litellm>=1.77.4",
+]