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

refactor: renamed agents to tools

Browse source
This commit is contained in:
Daniel Eder 2025-09-28 10:31:42 +02:00
parent 9b8584561e
commit ca493b8551
5 changed files with 126 additions and 121 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
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 automated "agents" alongside interactive play.
Python-based Telnet helper for connecting to MUD servers, handling login flows, and optionally running automated "tools" 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`, `MovementAgent`, `IntelligentCommunicationAgent`) with a pluggable interface for custom behaviours.
- Optional always-on tool mode plus an on-demand `#execute <tool>` escape hatch for ad-hoc automations.
- Built-in tools (`SimpleTool`, `ExploreTool`, `CommunicationTool`, `MovementTool`, `IntelligentCommunicationTool`) with a pluggable interface for custom behaviours.
## Requirements
@ -33,13 +33,13 @@ 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:
5. When *not* running in tool 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.
The command remains interactive while the tool works in the background and stops automatically a few seconds after things quiet down.
## Environment Variables
@ -53,37 +53,37 @@ 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 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`, `movement`, `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. |
| `MISTLE_TOOL_MODE` | ❌ | Enable full-time tool thread when set to truthy values (`1`, `true`, `yes`, `on`). Defaults to interactive-only mode. (`MISTLE_AGENT_MODE` still works as a backwards-compatible alias.) |
| `MISTLE_TOOL` | ❌ | Select which tool class to instantiate when tool mode is active. Accepted values: `simple` (default), `explore`, `communication`, `movement`, `intelligent`/`intelligentcommunication` (LLM-backed), or custom spec `module:ClassName`. (`MISTLE_AGENT` remains as a legacy alias.) |
| `MISTLE_LLM_MODEL` | ❌ | Override the `litellm` model used by the intelligent tool (defaults to `mistral/mistral-small-2407`). |
| `MISTRAL_API_KEY` | ❌ | API key used by `IntelligentCommunicationTool` (via `litellm`) when calling the `mistral/mistral-small-2407` model. |
## Agent Development
## Tool 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`.
- Implement new tools by subclassing `tools.Tool` and overriding `observe()` and `decide()`.
- Register the tool by either:
- Adding the class to `tools.py` and referencing it in `MISTLE_TOOL` (e.g., `explore` for `ExploreTool`).
- Placing the class elsewhere and configuring `MISTLE_TOOL` to `your_module:YourTool`.
- `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`.
- `MovementAgent` parses room descriptions/exits and issues direction commands (prefers `n`, `e`, `s`, `w`, diagonals, then vertical moves) while tracking which exits were already tried. Use `#execute move` for ad-hoc pathing or set `MISTLE_AGENT=movement` for continuous roaming.
- `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.
- Commands issued by the tool are throttled to one per second so manual commands can still interleave smoothly.
- `ExploreTool` showcases a richer workflow: it sends `schau`, identifies German nouns, inspects each with `untersuche`, and prints `[Tool]` progress updates like `Explored 3/7 — untersuche Tisch`.
- `MovementTool` parses room descriptions/exits and issues a single direction command, preferring unvisited exits and randomising choices to avoid oscillation. Trigger it via `#execute move` (or set `MISTLE_TOOL=movement` for continuous roaming).
- `CommunicationTool` auto-replies to every direct tell with a canned greeting, while `IntelligentCommunicationTool` routes each tell through `litellm` (default model `mistral/mistral-small-2407`) to craft a contextual answer via the configured LLM.
## On-Demand Agents
## On-Demand Tools
- 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 move`, `#execute intelligent`, or `#execute mypackage.mymodule:CustomAgent` are all valid.
- When `MISTLE_TOOL_MODE` is **off**, you can trigger an ephemeral tool at any time with `#execute <tool_spec>`.
- The syntax accepts the same values as `MISTLE_TOOL` and reuses the `build_tool` helper, so `#execute simple`, `#execute explore`, `#execute move`, `#execute intelligent`, or `#execute mypackage.mymodule:CustomTool` 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.
- When adding new tools, guard any long-running logic to avoid blocking the tool thread.
## Contributing
Feel free to open issues or submit pull requests for additional MUD-specific helpers, new agents, or quality-of-life improvements.
Feel free to open issues or submit pull requests for additional MUD-specific helpers, new tools, or quality-of-life improvements.
---

152
app.py
View file

@ -7,7 +7,7 @@ from pathlib import Path
from threading import Event, Lock, Thread
from typing import Callable, Optional, Type
from agent import Agent, SimpleAgent
from tools import Tool, SimpleTool
from telnetclient import TelnetClient
@ -19,13 +19,13 @@ class SessionState:
self._output_lock = Lock()
self._output_event = Event()
self._last_output = ""
self._last_agent_send = 0.0
self._last_tool_send = 0.0
def send(self, client: TelnetClient, message: str) -> None:
with self._send_lock:
client.send(message)
def agent_send(
def tool_send(
self,
client: TelnetClient,
message: str,
@ -33,14 +33,14 @@ class SessionState:
min_interval: float,
stop_event: Event,
) -> bool:
"""Send on behalf of the agent while respecting a minimum cadence."""
"""Send on behalf of the tool while respecting a minimum cadence."""
while not stop_event.is_set():
with self._send_lock:
now = time.time()
elapsed = now - self._last_agent_send
elapsed = now - self._last_tool_send
if elapsed >= min_interval:
client.send(message)
self._last_agent_send = now
self._last_tool_send = now
return True
wait_time = min_interval - elapsed
if wait_time <= 0:
@ -67,10 +67,10 @@ class SessionState:
self._output_event.clear()
def run_agent_loop(
def run_tool_loop(
client: TelnetClient,
state: SessionState,
agent: Agent,
tool: Tool,
stop_event: Event,
*,
idle_delay: float = 0.5,
@ -78,19 +78,19 @@ def run_agent_loop(
auto_stop: bool = False,
auto_stop_idle: float = 2.0,
) -> None:
"""Invoke *agent* whenever new output arrives and send its response."""
"""Invoke *tool* whenever new output arrives and send its response."""
idle_started: Optional[float] = None
def maybe_send() -> None:
nonlocal idle_started
try:
command = agent.decide()
command = tool.decide()
except Exception as exc: # pragma: no cover - defensive logging
print(f"Agent failed: {exc}", file=sys.stderr)
print(f"[Tool] Failed: {exc}", file=sys.stderr)
return
if not command:
return
sent = state.agent_send(
sent = state.tool_send(
client,
command,
min_interval=min_send_interval,
@ -120,9 +120,9 @@ def run_agent_loop(
if not last_output:
continue
try:
agent.observe(last_output)
tool.observe(last_output)
except Exception as exc: # pragma: no cover - defensive logging
print(f"Agent failed during observe: {exc}", file=sys.stderr)
print(f"[Tool] Failed during observe: {exc}", file=sys.stderr)
continue
maybe_send()
@ -152,67 +152,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()
def build_tool(spec: str) -> Tool:
"""Instantiate a tool based on configuration."""
normalized = spec.strip()
if not normalized:
return SimpleAgent()
return SimpleTool()
key = normalized.lower()
if key == "simple":
return SimpleAgent()
return SimpleTool()
builtin_agents = {
"explore": ("agent", "ExploreAgent", {}),
"communication": ("agent", "CommunicationAgent", {}),
"movement": ("movement_agent", "MovementAgent", {}),
"move": ("movement_agent", "MovementAgent", {}),
builtin_tools = {
"explore": ("tools", "ExploreTool", {}),
"communication": ("tools", "CommunicationTool", {}),
"movement": ("movement_tool", "MovementTool", {}),
"move": ("movement_tool", "MovementTool", {}),
"intelligent": (
"intelligent_agent",
"IntelligentCommunicationAgent",
"intelligent_tool",
"IntelligentCommunicationTool",
{"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407")},
),
"intelligentcommunication": (
"intelligent_agent",
"IntelligentCommunicationAgent",
"intelligent_tool",
"IntelligentCommunicationTool",
{"model": os.environ.get("MISTLE_LLM_MODEL", "mistral/mistral-small-2407")},
),
}
if key in builtin_agents:
module_name, class_name, kwargs = builtin_agents[key]
if key in builtin_tools:
module_name, class_name, kwargs = builtin_tools[key]
try:
module = import_module(module_name)
agent_cls = getattr(module, class_name)
tool_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)
raise RuntimeError(f"{class_name} is not available in tools module") from exc
tool = _instantiate_tool(tool_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
print(f"[Tool] Using LLM model: {model_name}")
return tool
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")
raise RuntimeError("MISTLE_TOOL must be in 'module:ClassName' format")
module = import_module(module_name)
agent_cls = getattr(module, class_name)
return _instantiate_agent(agent_cls, normalized)
tool_cls = getattr(module, class_name)
return _instantiate_tool(tool_cls, normalized)
raise RuntimeError(f"Unknown agent spec '{agent_spec}'.")
raise RuntimeError(f"Unknown tool spec '{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")
def _instantiate_tool(
tool_cls: Type[Tool], tool_spec: str, kwargs: Optional[dict] = None
) -> Tool:
if not issubclass(tool_cls, Tool):
raise RuntimeError(f"{tool_spec} is not a Tool subclass")
try:
kwargs = kwargs or {}
return agent_cls(**kwargs)
return tool_cls(**kwargs)
except TypeError as exc:
raise RuntimeError(f"Failed to instantiate {agent_spec}: {exc}") from exc
raise RuntimeError(f"Failed to instantiate {tool_spec}: {exc}") from exc
def login(
@ -256,7 +256,7 @@ def interactive_session(
poll_interval: float = 0.2,
receive_timeout: float = 0.2,
exit_command: str,
agent_command: Optional[Callable[[str], None]] = None,
tool_command: Optional[Callable[[str], None]] = None,
) -> None:
"""Keep the Telnet session running, proxying input/output until interrupted."""
if exit_command:
@ -279,12 +279,12 @@ def interactive_session(
line = line.rstrip("\r\n")
if not line:
continue
if agent_command and line.lower().startswith("#execute"):
if tool_command and line.lower().startswith("#execute"):
parts = line.split(maxsplit=1)
if len(parts) == 1:
print("[Agent] Usage: #execute <agent_spec>")
print("[Tool] Usage: #execute <tool_spec>")
else:
agent_command(parts[1])
tool_command(parts[1])
continue
state.send(client, line)
@ -326,14 +326,20 @@ def main() -> int:
password = os.environ.get("MISTLE_PASSWORD", "")
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", "")
tool_mode_env = os.environ.get("MISTLE_TOOL_MODE")
if tool_mode_env is None:
tool_mode_env = os.environ.get("MISTLE_AGENT_MODE", "")
tool_mode = tool_mode_env.lower() in {"1", "true", "yes", "on"}
tool_spec = os.environ.get("MISTLE_TOOL")
if tool_spec is None:
tool_spec = os.environ.get("MISTLE_AGENT", "")
state = SessionState()
stop_event = Event()
agent_thread: Optional[Thread] = None
agent: Optional[Agent] = None
ephemeral_agents: list[Thread] = []
tool_thread: Optional[Thread] = None
tool: Optional[Tool] = None
ephemeral_tools: list[Thread] = []
with TelnetClient(host=host, port=port, timeout=10.0) as client:
login(
@ -344,29 +350,29 @@ def main() -> int:
state=state,
)
if agent_mode:
agent = build_agent(agent_spec)
agent_thread = Thread(
target=run_agent_loop,
args=(client, state, agent, stop_event),
if tool_mode:
tool = build_tool(tool_spec)
tool_thread = Thread(
target=run_tool_loop,
args=(client, state, tool, stop_event),
kwargs={"min_send_interval": 1.0},
daemon=True,
)
agent_thread.start()
tool_thread.start()
def run_ephemeral_agent(spec: str) -> None:
def run_ephemeral_tool(spec: str) -> None:
spec = spec.strip()
if not spec:
print("[Agent] Usage: #execute <agent_spec>")
print("[Tool] Usage: #execute <tool_spec>")
return
try:
temp_agent = build_agent(spec)
temp_tool = build_tool(spec)
except RuntimeError as exc:
print(f"[Agent] Failed to load '{spec}': {exc}", file=sys.stderr)
print(f"[Tool] Failed to load '{spec}': {exc}", file=sys.stderr)
return
thread = Thread(
target=run_agent_loop,
args=(client, state, temp_agent, stop_event),
target=run_tool_loop,
args=(client, state, temp_tool, stop_event),
kwargs={
"min_send_interval": 1.0,
"auto_stop": True,
@ -374,8 +380,8 @@ def main() -> int:
},
daemon=True,
)
ephemeral_agents.append(thread)
print(f"[Agent] Executing {spec!r} once")
ephemeral_tools.append(thread)
print(f"[Tool] Executing {spec!r} once")
thread.start()
interrupted = False
@ -385,16 +391,16 @@ def main() -> int:
state=state,
stop_event=stop_event,
exit_command=exit_command,
agent_command=None if agent_mode else run_ephemeral_agent,
tool_command=None if tool_mode else run_ephemeral_tool,
)
except KeyboardInterrupt:
print()
interrupted = True
finally:
stop_event.set()
if agent_thread:
agent_thread.join(timeout=1.0)
for thread in ephemeral_agents:
if tool_thread:
tool_thread.join(timeout=1.0)
for thread in ephemeral_tools:
thread.join(timeout=1.0)
if interrupted:

14
intelligent_agent.py → intelligent_tool.py
View file

@ -11,12 +11,12 @@ try:
except ImportError: # pragma: no cover - optional dependency
completion = None # type: ignore[assignment]
from agent import Agent
from tools import Tool
@dataclass
class IntelligentCommunicationAgent(Agent):
"""Agent that uses a language model to answer private tells."""
class IntelligentCommunicationTool(Tool):
"""Tool that uses a language model to answer private tells."""
model: str = "mistral/mistral-tiny"
system_prompt: str = (
@ -47,7 +47,7 @@ class IntelligentCommunicationAgent(Agent):
if not player:
continue
self.pending_replies.append((player, message))
print(f"[Agent] Received message from {player}: {message}")
print(f"[Tool] Received message from {player}: {message}")
self._append_history(player, "user", message)
def decide(self) -> Optional[str]:
@ -57,13 +57,13 @@ class IntelligentCommunicationAgent(Agent):
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")
print(f"[Tool] 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",
"[Tool] litellm is not installed; falling back to default reply",
file=sys.stderr,
)
return self.fallback_reply
@ -76,7 +76,7 @@ class IntelligentCommunicationAgent(Agent):
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)
print(f"[Tool] Model call failed: {exc}", file=sys.stderr)
return self.fallback_reply
try:

10
movement_agent.py → movement_tool.py
View file

@ -5,12 +5,12 @@ import random
from dataclasses import dataclass, field
from typing import Dict, Optional, Pattern, Sequence, Set, Tuple
from agent import Agent
from tools import Tool
@dataclass
class MovementAgent(Agent):
"""Agent that chooses movement commands based on room descriptions."""
class MovementTool(Tool):
"""Tool that issues a single movement based on the room description."""
look_command: str = "schau"
command_format: str = "{direction}"
@ -102,7 +102,7 @@ class MovementAgent(Agent):
return None
if self.needs_look:
self.needs_look = False
print("[Agent] Requesting room description via look command")
print("[Tool] Requesting room description via look command")
return self.look_command
if self.selected_direction is None:
return None
@ -112,7 +112,7 @@ class MovementAgent(Agent):
direction
)
command = self.command_format.format(direction=direction)
print(f"[Agent] Moving via {direction}")
print(f"[Tool] Moving via {direction}")
self.executed = True
self.selected_direction = None
return command

25
agent.py → tools.py
View file

@ -8,8 +8,8 @@ from dataclasses import dataclass, field
from typing import Deque, Optional, Pattern, Set, Tuple
class Agent(ABC):
"""Interface for autonomous Telnet actors."""
class Tool(ABC):
"""Interface for autonomous Telnet tools."""
@abstractmethod
def observe(self, output: str) -> None:
@ -21,8 +21,8 @@ class Agent(ABC):
@dataclass
class SimpleAgent(Agent):
"""Minimal agent that always returns the same command."""
class SimpleTool(Tool):
"""Minimal tool that always returns the same command."""
default_command: str = "schau"
last_output: str = field(default="", init=False)
@ -36,8 +36,8 @@ class SimpleAgent(Agent):
@dataclass
class ExploreAgent(Agent):
"""Agent that inspects every noun it discovers in the room description."""
class ExploreTool(Tool):
"""Tool that inspects every noun it discovers in the room description."""
look_command: str = "schau"
inspect_command: str = "untersuche"
@ -68,14 +68,14 @@ class ExploreAgent(Agent):
def decide(self) -> Optional[str]:
if not self.look_sent:
self.look_sent = True
print("[Agent] Exploring room, sending 'schau'")
print("[Tool] 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}"
progress = f"[Tool] Explored {len(self.inspected_targets)}/{len(self.seen_targets)} — untersuche {target}"
print(progress)
return f"{self.inspect_command} {target}"
@ -83,8 +83,8 @@ class ExploreAgent(Agent):
@dataclass
class CommunicationAgent(Agent):
"""Agent that replies to private tells."""
class CommunicationTool(Tool):
"""Tool that replies to private tells."""
reply_template: str = "teile {player} mit Hallo! Ich bin Mistle und ein Bot."
tell_pattern: Pattern[str] = field(
@ -106,13 +106,12 @@ class CommunicationAgent(Agent):
if not player:
continue
self.pending_replies.append((player, message))
print(f"[Agent] Received message from {player}: {message}")
print(f"[Tool] 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}")
print(f"[Tool] Replying to {player}")
return reply