feat: fixed startegy agent

README.md

@@ -8,6 +8,7 @@ Python-based Telnet helper for connecting to MUD servers, handling login flows,
 - 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 tool mode plus an on-demand `#execute <tool>` escape hatch for ad-hoc automations.
+- Higher-level agents (`FixedStrategyAgent` so far) that can string multiple tools together via `#agent <spec>`.
 - Built-in tools (`SimpleTool`, `ExploreTool`, `CommunicationTool`, `MovementTool`, `IntelligentCommunicationTool`) with a pluggable interface for custom behaviours.
 
 ## Requirements
@@ -41,6 +42,14 @@ Python-based Telnet helper for connecting to MUD servers, handling login flows,
 
    The command remains interactive while the tool works in the background and stops automatically a few seconds after things quiet down.
 
+6. To run an agent that orchestrates several tools, use:
+
+   ```text
+   #agent move,explore
+   ```
+
+   This example uses the fixed-strategy agent to run `move` and then `explore` once.
+
 ## Environment Variables
 
 All variables can be placed in the `.env` file (one `KEY=value` per line) or provided through the shell environment.

agent_runtime.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import sys
+from threading import Event
+from typing import Callable
+
+from agents import Agent
+from tools import Tool
+
+ToolBuilder = Callable[[str], Tool]
+ToolRunner = Callable[[Tool], None]
+
+
+def run_agent(
+    agent: Agent,
+    *,
+    build_tool: ToolBuilder,
+    run_tool: ToolRunner,
+    stop_event: Event,
+) -> None:
+    """Execute *agent* by wiring it to the tool runtime."""
+
+    def invoke_tool(spec: str) -> bool:
+        name = spec.strip()
+        if not name:
+            print("[Agent] Ignoring empty tool spec", file=sys.stderr)
+            return False
+        try:
+            tool = build_tool(name)
+        except RuntimeError as exc:
+            print(f"[Agent] Failed to load tool '{name}': {exc}", file=sys.stderr)
+            return False
+        print(f"[Agent] Running tool '{name}'")
+        run_tool(tool)
+        return True
+
+    agent.run(invoke_tool=invoke_tool, stop_event=stop_event)

agents.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import sys
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from threading import Event
+from typing import Callable, Iterable
+
+
+ToolInvoker = Callable[[str], bool]
+
+
+class Agent(ABC):
+    """Interface for higher-level behaviours that orchestrate tools."""
+
+    @abstractmethod
+    def run(self, *, invoke_tool: ToolInvoker, stop_event: Event) -> None:
+        """Execute the agent strategy until finished or *stop_event* is set."""
+
+
+@dataclass
+class FixedStrategyAgent(Agent):
+    """Run a comma-separated list of tool specs in order."""
+
+    plan: str
+
+    def run(self, *, invoke_tool: ToolInvoker, stop_event: Event) -> None:
+        steps = [part.strip() for part in self.plan.split(",") if part.strip()]
+        if not steps:
+            print("[Agent] No tools configured for fixed strategy", file=sys.stderr)
+            return
+
+        for step in steps:
+            if stop_event.is_set():
+                break
+            success = invoke_tool(step)
+            if not success:
+                # Abort on first failure to avoid confusing behaviour
+                break
+
+
+def build_agent(spec: str) -> Agent:
+    normalized = spec.strip()
+    if not normalized:
+        raise RuntimeError("Agent specification must not be empty")
+
+    if ":" in normalized:
+        kind, config = normalized.split(":", 1)
+    else:
+        kind, config = "fixed", normalized
+
+    kind = kind.strip().lower()
+    config = config.strip()
+
+    if kind in {"fixed", "strategy", "fixedstrategy"}:
+        return FixedStrategyAgent(config)
+
+    raise RuntimeError(f"Unknown agent type '{kind}'")

app.py

@@ -8,6 +8,8 @@ from threading import Event, Lock, Thread
 from typing import Callable, Optional, Type
 
 from tools import Tool, SimpleTool
+from agents import Agent, build_agent
+from agent_runtime import run_agent
 from telnetclient import TelnetClient
 
 
@@ -126,7 +128,6 @@ def run_tool_loop(
             continue
         maybe_send()
 
-
 def load_env_file(path: str = ".env") -> None:
     """Populate ``os.environ`` with key/value pairs from a dotenv file."""
     env_path = Path(path)
@@ -257,6 +258,7 @@ def interactive_session(
     receive_timeout: float = 0.2,
     exit_command: str,
     tool_command: Optional[Callable[[str], None]] = None,
+    agent_command: Optional[Callable[[str], None]] = None,
 ) -> None:
     """Keep the Telnet session running, proxying input/output until interrupted."""
     if exit_command:
@@ -279,7 +281,15 @@ def interactive_session(
             line = line.rstrip("\r\n")
             if not line:
                 continue
-            if tool_command and line.lower().startswith("#execute"):
+            lowered = line.lower()
+            if agent_command and lowered.startswith("#agent"):
+                parts = line.split(maxsplit=1)
+                if len(parts) == 1:
+                    print("[Agent] Usage: #agent <agent_spec>")
+                else:
+                    agent_command(parts[1])
+                continue
+            if tool_command and lowered.startswith("#execute"):
                 parts = line.split(maxsplit=1)
                 if len(parts) == 1:
                     print("[Tool] Usage: #execute <tool_spec>")
@@ -336,6 +346,7 @@ def main() -> int:
     tool_thread: Optional[Thread] = None
     tool: Optional[Tool] = None
     ephemeral_tools: list[Thread] = []
+    agent_threads: list[Thread] = []
 
     with TelnetClient(host=host, port=port, timeout=10.0) as client:
         login(
@@ -380,6 +391,42 @@ def main() -> int:
             print(f"[Tool] Executing {spec!r} once")
             thread.start()
 
+        def run_ephemeral_agent(spec: str) -> None:
+            spec = spec.strip()
+            if not spec:
+                print("[Agent] Usage: #agent <agent_spec>")
+                return
+            try:
+                temp_agent = build_agent(spec)
+            except RuntimeError as exc:
+                print(f"[Agent] Failed to configure '{spec}': {exc}", file=sys.stderr)
+                return
+
+            def run_tool_instance(tool: Tool) -> None:
+                run_tool_loop(
+                    client,
+                    state,
+                    tool,
+                    stop_event,
+                    min_send_interval=1.0,
+                    auto_stop=True,
+                    auto_stop_idle=2.0,
+                )
+
+            thread = Thread(
+                target=run_agent,
+                args=(temp_agent,),
+                kwargs={
+                    "build_tool": build_tool,
+                    "run_tool": run_tool_instance,
+                    "stop_event": stop_event,
+                },
+                daemon=True,
+            )
+            agent_threads.append(thread)
+            print(f"[Agent] Executing {spec!r}")
+            thread.start()
+
         interrupted = False
         try:
             interactive_session(
@@ -388,6 +435,7 @@ def main() -> int:
                 stop_event=stop_event,
                 exit_command=exit_command,
                 tool_command=None if tool_mode else run_ephemeral_tool,
+                agent_command=run_ephemeral_agent,
             )
         except KeyboardInterrupt:
             print()
@@ -398,6 +446,8 @@ def main() -> int:
                 tool_thread.join(timeout=1.0)
             for thread in ephemeral_tools:
                 thread.join(timeout=1.0)
+            for thread in agent_threads:
+                thread.join(timeout=1.0)
 
         if interrupted:
             graceful_shutdown(client, exit_command, state=state)