DevZoneX commited on
Commit
7ec4d32
·
1 Parent(s): 7a36b3c

Implement ReAct agent and MCP tools using the example submission

Browse files
Files changed (2) hide show
  1. agent.py +275 -148
  2. mcp_server.py +143 -156
agent.py CHANGED
@@ -1,26 +1,8 @@
1
  """
2
- Student Agent for Text Adventure Games
3
 
4
- This is your submission file. Implement the StudentAgent class to play
5
- text adventure games using the MCP server you also implement.
6
-
7
- Your agent should:
8
- 1. Connect to the MCP server via the provided client
9
- 2. Use the ReAct pattern (Thought -> Action -> Observation)
10
- 3. Call MCP tools to interact with the game
11
- 4. Maximize the game score within the step limit
12
-
13
- Required method:
14
- async def run(self, client, game, max_steps, seed, verbose) -> RunResult
15
-
16
- The 'client' is a FastMCP Client already connected to your MCP server.
17
- Use it to call tools like: await client.call_tool("play_action", {"action": "look"})
18
-
19
- Tips:
20
- - Start by looking around and understanding your environment
21
- - Keep track of visited locations to avoid loops
22
- - Pick up useful items (lamp, sword, etc.)
23
- - The seed parameter should be used to set your LLM's seed for reproducibility
24
  """
25
 
26
  import json
@@ -32,17 +14,14 @@ from typing import Optional
32
  from dotenv import load_dotenv
33
  from huggingface_hub import InferenceClient
34
 
35
- # Load environment variables
36
  load_dotenv()
37
 
38
  # =============================================================================
39
  # LLM Configuration - DO NOT MODIFY
40
  # =============================================================================
41
 
42
- # Model to use (fixed for fair evaluation)
43
  LLM_MODEL = "Qwen/Qwen2.5-72B-Instruct"
44
 
45
- # Initialize the LLM client (uses HF_TOKEN from environment)
46
  _hf_token = os.getenv("HF_TOKEN")
47
  if not _hf_token:
48
  raise ValueError("HF_TOKEN not found. Set it in your .env file.")
@@ -51,25 +30,7 @@ LLM_CLIENT = InferenceClient(token=_hf_token)
51
 
52
 
53
  def call_llm(prompt: str, system_prompt: str, seed: int, max_tokens: int = 300) -> str:
54
- """
55
- Call the LLM with the given prompt. Use this function in your agent.
56
-
57
- Args:
58
- prompt: The user prompt (current game state, history, etc.)
59
- system_prompt: The system prompt (instructions for the agent)
60
- seed: Random seed for reproducibility
61
- max_tokens: Maximum tokens in response (default: 300)
62
-
63
- Returns:
64
- The LLM's response text
65
-
66
- Example:
67
- response = call_llm(
68
- prompt="You are in a forest. What do you do?",
69
- system_prompt=SYSTEM_PROMPT,
70
- seed=42,
71
- )
72
- """
73
  messages = [
74
  {"role": "system", "content": system_prompt},
75
  {"role": "user", "content": prompt},
@@ -78,7 +39,7 @@ def call_llm(prompt: str, system_prompt: str, seed: int, max_tokens: int = 300)
78
  response = LLM_CLIENT.chat.completions.create(
79
  model=LLM_MODEL,
80
  messages=messages,
81
- temperature=0.0, # Deterministic for reproducibility
82
  max_tokens=max_tokens,
83
  seed=seed,
84
  )
@@ -99,179 +60,345 @@ class RunResult:
99
 
100
 
101
  # =============================================================================
102
- # System Prompt - Customize this for your agent
103
  # =============================================================================
104
 
105
- SYSTEM_PROMPT = """You are playing a classic text adventure game.
106
-
107
- GOAL: Explore the world, solve puzzles, and maximize your score.
108
 
109
- AVAILABLE TOOLS (use via MCP):
110
- - play_action: Execute a game command (north, take lamp, open mailbox, etc.)
111
- - memory: Get current game state and history (if implemented)
112
- - inventory: Check what you're carrying (if implemented)
 
113
 
114
  VALID GAME COMMANDS for play_action:
115
  - Movement: north, south, east, west, up, down, enter, exit
116
  - Objects: take <item>, drop <item>, open <thing>, close <thing>, examine <thing>
117
- - Other: look, inventory, read <thing>, turn on lamp
 
 
 
 
118
 
119
  RESPOND IN THIS EXACT FORMAT (no markdown):
120
- THOUGHT: <your reasoning about what to do next>
121
  TOOL: <tool_name>
122
- ARGS: <JSON arguments, e.g., {"action": "look"}>
123
 
124
- Example:
125
- THOUGHT: I should look around to see where I am.
126
  TOOL: play_action
127
  ARGS: {"action": "look"}
128
- """
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
129
 
130
 
131
  # =============================================================================
132
- # Student Agent - IMPLEMENT THIS CLASS
133
  # =============================================================================
134
 
135
  class StudentAgent:
136
  """
137
- Your ReAct agent implementation.
138
 
139
- TODO:
140
- 1. Implement the run() method with the ReAct loop
141
- 2. Parse LLM responses to extract tool calls
142
- 3. Track state and avoid loops
143
-
144
- Use the provided call_llm() function to interact with the LLM.
145
  """
146
 
147
  def __init__(self):
148
- """Initialize your agent here."""
149
- # TODO: Initialize any state tracking you need
150
- # self.history = []
151
- # self.visited_locations = set()
152
- pass
153
 
154
  async def run(
155
  self,
156
- client, # FastMCP Client connected to your MCP server
157
  game: str,
158
  max_steps: int,
159
  seed: int,
160
  verbose: bool = False,
161
  ) -> RunResult:
162
- """
163
- Run the agent for a game session.
 
 
164
 
165
- Args:
166
- client: FastMCP Client connected to your MCP server
167
- game: Name of the game being played (e.g., "zork1")
168
- max_steps: Maximum number of steps to take
169
- seed: Random seed for reproducibility (use for LLM calls)
170
- verbose: Whether to print detailed output
171
-
172
- Returns:
173
- RunResult with final score and statistics
174
- """
175
- # TODO: Implement your ReAct loop here
176
- #
177
- # Basic structure:
178
- # 1. Get initial observation (call play_action with "look")
179
- # 2. Loop for max_steps:
180
- # a. Build prompt with current observation and history
181
- # b. Call LLM to get thought and action
182
- # c. Parse the response to extract tool and args
183
- # d. Call the tool via client.call_tool(tool_name, args)
184
- # e. Update history and state
185
- # f. Check for game over
186
- # 3. Return RunResult with final statistics
187
 
188
- # Example of calling a tool:
189
- # result = await client.call_tool("play_action", {"action": "look"})
190
- # observation = result[0].text if result else "No response"
191
 
192
- # Example of calling the LLM:
193
- # response = call_llm(
194
- # prompt="Current observation: " + observation,
195
- # system_prompt=SYSTEM_PROMPT,
196
- # seed=seed,
197
- # )
198
 
199
- # Placeholder implementation - replace with your code
200
- locations_visited = set()
201
- history = []
202
- final_score = 0
203
- moves = 0
204
 
205
- # TODO: Your implementation here
206
- # ...
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
207
 
208
  return RunResult(
209
- final_score=final_score,
210
- max_score=350, # Zork1 max score, adjust if needed
211
  moves=moves,
212
  locations_visited=locations_visited,
213
- game_completed=False,
214
  history=history,
215
  )
216
 
217
- def _build_prompt(self, observation: str, history: list) -> str:
218
- """
219
- Build the prompt for the LLM.
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
220
 
221
- TODO: Implement this to create effective prompts
222
- """
223
- # TODO: Combine system prompt, history, and current observation
224
- pass
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
225
 
226
- def _parse_response(self, response: str) -> tuple[str, str, dict]:
227
- """
228
- Parse LLM response to extract thought, tool name, and arguments.
 
 
 
 
 
 
 
 
 
 
 
229
 
230
- TODO: Implement robust parsing
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
231
 
232
- Returns:
233
- Tuple of (thought, tool_name, args_dict)
234
- """
235
- # TODO: Parse the response format:
236
- # THOUGHT: ...
237
- # TOOL: ...
238
- # ARGS: {...}
239
- pass
 
240
 
241
- def _call_llm(self, prompt: str, system_prompt: str, seed: int) -> str:
242
- """
243
- Call the LLM with the given prompt.
 
 
 
 
244
 
245
- This is a convenience wrapper - you can also use call_llm() directly.
246
- """
247
- return call_llm(prompt, system_prompt, seed)
 
 
 
 
 
 
 
 
 
 
 
 
248
 
249
 
250
  # =============================================================================
251
- # For local testing
252
  # =============================================================================
253
 
254
  async def test_agent():
255
  """Test the agent locally."""
256
  from fastmcp import Client
257
 
258
- # Path to your MCP server
259
- server_path = "mcp_server.py"
260
-
261
  agent = StudentAgent()
 
 
262
 
263
- async with Client(server_path) as client:
264
  result = await agent.run(
265
  client=client,
266
  game="zork1",
267
- max_steps=10,
268
  seed=42,
269
  verbose=True,
270
  )
271
 
272
- print(f"\nFinal Score: {result.final_score}")
 
273
  print(f"Moves: {result.moves}")
274
- print(f"Locations: {result.locations_visited}")
275
 
276
 
277
  if __name__ == "__main__":
 
1
  """
2
+ Example: MCP ReAct Agent
3
 
4
+ A complete ReAct agent that uses MCP tools to play text adventure games.
5
+ This is a working example students can learn from.
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
6
  """
7
 
8
  import json
 
14
  from dotenv import load_dotenv
15
  from huggingface_hub import InferenceClient
16
 
 
17
  load_dotenv()
18
 
19
  # =============================================================================
20
  # LLM Configuration - DO NOT MODIFY
21
  # =============================================================================
22
 
 
23
  LLM_MODEL = "Qwen/Qwen2.5-72B-Instruct"
24
 
 
25
  _hf_token = os.getenv("HF_TOKEN")
26
  if not _hf_token:
27
  raise ValueError("HF_TOKEN not found. Set it in your .env file.")
 
30
 
31
 
32
  def call_llm(prompt: str, system_prompt: str, seed: int, max_tokens: int = 300) -> str:
33
+ """Call the LLM with the given prompt."""
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
34
  messages = [
35
  {"role": "system", "content": system_prompt},
36
  {"role": "user", "content": prompt},
 
39
  response = LLM_CLIENT.chat.completions.create(
40
  model=LLM_MODEL,
41
  messages=messages,
42
+ temperature=0.0,
43
  max_tokens=max_tokens,
44
  seed=seed,
45
  )
 
60
 
61
 
62
  # =============================================================================
63
+ # System Prompt
64
  # =============================================================================
65
 
66
+ SYSTEM_PROMPT = """You are an expert text adventure game player. Your goal is to explore, collect treasures, and maximize your score.
 
 
67
 
68
+ AVAILABLE TOOLS (use these via MCP):
69
+ 1. play_action - Execute game commands (north, take lamp, open mailbox, etc.)
70
+ 2. memory - Get current game state, score, and recent history
71
+ 3. get_map - See explored locations and connections
72
+ 4. inventory - Check what you're carrying
73
 
74
  VALID GAME COMMANDS for play_action:
75
  - Movement: north, south, east, west, up, down, enter, exit
76
  - Objects: take <item>, drop <item>, open <thing>, close <thing>, examine <thing>
77
+ - Light: turn on lamp, turn off lamp
78
+ - Combat: attack <enemy> with <weapon>
79
+ - Other: inventory, look, read <thing>, wait
80
+
81
+ FORBIDDEN (will NOT work): check, inspect, search, grab, use, help
82
 
83
  RESPOND IN THIS EXACT FORMAT (no markdown):
84
+ THOUGHT: <brief reasoning about what to do next>
85
  TOOL: <tool_name>
86
+ ARGS: <JSON arguments>
87
 
88
+ Examples:
89
+ THOUGHT: I need to see what's around me.
90
  TOOL: play_action
91
  ARGS: {"action": "look"}
92
+
93
+ THOUGHT: Let me check my current state and score.
94
+ TOOL: memory
95
+ ARGS: {}
96
+
97
+ THOUGHT: The mailbox might contain something useful.
98
+ TOOL: play_action
99
+ ARGS: {"action": "open mailbox"}
100
+
101
+ STRATEGY:
102
+ 1. Start by looking around and checking memory
103
+ 2. Explore systematically - try all directions
104
+ 3. Pick up useful items (lamp, sword, etc.)
105
+ 4. Open containers (mailbox, window, etc.)
106
+ 5. Use get_map to avoid getting lost
107
+ 6. Turn on lamp before dark areas!
108
+
109
+ DO NOT repeat the same action multiple times in a row."""
110
 
111
 
112
  # =============================================================================
113
+ # Student Agent Implementation
114
  # =============================================================================
115
 
116
  class StudentAgent:
117
  """
118
+ MCP ReAct Agent - A complete working example.
119
 
120
+ This agent demonstrates:
121
+ - ReAct loop (Thought -> Tool -> Observation)
122
+ - Loop detection
123
+ - Action validation
124
+ - Score tracking via memory tool
 
125
  """
126
 
127
  def __init__(self):
128
+ """Initialize the agent state."""
129
+ self.history: list[dict] = []
130
+ self.recent_actions: list[str] = []
131
+ self.score: int = 0
 
132
 
133
  async def run(
134
  self,
135
+ client,
136
  game: str,
137
  max_steps: int,
138
  seed: int,
139
  verbose: bool = False,
140
  ) -> RunResult:
141
+ """Run the agent for a game session."""
142
+ locations_visited = set()
143
+ history = []
144
+ moves = 0
145
 
146
+ # Get list of available tools
147
+ tools = await client.list_tools()
148
+ tool_names = [t.name for t in tools]
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
149
 
150
+ # Get initial observation
151
+ result = await client.call_tool("play_action", {"action": "look"})
152
+ observation = self._extract_result(result)
153
 
154
+ # Track initial location
155
+ location = observation.split("\n")[0] if observation else "Unknown"
156
+ locations_visited.add(location)
 
 
 
157
 
158
+ if verbose:
159
+ print(f"\n{observation}")
 
 
 
160
 
161
+ # Main ReAct loop
162
+ for step in range(1, max_steps + 1):
163
+ # Build prompt with context
164
+ prompt = self._build_prompt(observation)
165
+
166
+ # Call LLM for reasoning (use step-based seed for variety)
167
+ response = call_llm(prompt, SYSTEM_PROMPT, seed + step)
168
+
169
+ # Parse the response
170
+ thought, tool_name, tool_args = self._parse_response(response, tool_names)
171
+
172
+ if verbose:
173
+ print(f"\n--- Step {step} ---")
174
+ print(f"[THOUGHT] {thought}")
175
+ print(f"[TOOL] {tool_name}({tool_args})")
176
+
177
+ # Validate and fix common issues
178
+ tool_name, tool_args = self._validate_tool_call(tool_name, tool_args, tool_names)
179
+
180
+ # Loop detection
181
+ if tool_name == "play_action":
182
+ action = tool_args.get("action", "look")
183
+ self.recent_actions.append(action)
184
+ if len(self.recent_actions) > 5:
185
+ self.recent_actions = self.recent_actions[-5:]
186
+
187
+ # Detect loops - if same action 3 times, force "look"
188
+ if len(self.recent_actions) >= 3 and len(set(self.recent_actions[-3:])) == 1:
189
+ if verbose:
190
+ print(f"[WARNING] Loop detected - forcing 'look'")
191
+ tool_args = {"action": "look"}
192
+ self.recent_actions.append("look")
193
+
194
+ moves += 1
195
+
196
+ # Execute the tool
197
+ try:
198
+ result = await client.call_tool(tool_name, tool_args)
199
+ observation = self._extract_result(result)
200
+
201
+ if verbose:
202
+ print(f"[RESULT] {observation[:200]}...")
203
+ except Exception as e:
204
+ observation = f"Error: {e}"
205
+ if verbose:
206
+ print(f"[ERROR] {e}")
207
+
208
+ # Track location
209
+ location = observation.split("\n")[0] if observation else "Unknown"
210
+ locations_visited.add(location)
211
+
212
+ # Update history
213
+ self.history.append({
214
+ "step": step,
215
+ "thought": thought,
216
+ "tool": tool_name,
217
+ "args": tool_args,
218
+ "result": observation[:200]
219
+ })
220
+ if len(self.history) > 10:
221
+ self.history = self.history[-10:]
222
+
223
+ # Track score from observation
224
+ self._update_score(observation)
225
+
226
+ # Record in result history
227
+ history.append((thought, f"{tool_name}({tool_args})", observation[:100]))
228
+
229
+ # Check for game over
230
+ if self._is_game_over(observation):
231
+ if verbose:
232
+ print("\n*** GAME OVER ***")
233
+ break
234
 
235
  return RunResult(
236
+ final_score=self.score,
237
+ max_score=350,
238
  moves=moves,
239
  locations_visited=locations_visited,
240
+ game_completed=self._is_game_over(observation),
241
  history=history,
242
  )
243
 
244
+ def _build_prompt(self, observation: str) -> str:
245
+ """Build the prompt for the LLM with context."""
246
+ parts = []
247
+
248
+ parts.append(f"Current Score: {self.score}")
249
+
250
+ # Recent history
251
+ if self.history:
252
+ parts.append("\nRecent actions:")
253
+ for entry in self.history[-3:]:
254
+ action = entry.get("args", {}).get("action", entry["tool"])
255
+ result_short = entry["result"][:80] + "..." if len(entry["result"]) > 80 else entry["result"]
256
+ parts.append(f" > {action} -> {result_short}")
257
+
258
+ # Warn about repeated actions
259
+ if self.recent_actions and len(set(self.recent_actions[-3:])) == 1:
260
+ parts.append(f"\n[WARNING: You've been doing '{self.recent_actions[-1]}' repeatedly. TRY SOMETHING DIFFERENT!]")
261
+
262
+ parts.append(f"\nCurrent situation:\n{observation}")
263
+ parts.append("\nWhat do you do next?")
264
+
265
+ return "\n".join(parts)
266
+
267
+ def _parse_response(self, response: str, valid_tools: list[str]) -> tuple[str, str, dict]:
268
+ """Parse the LLM response to extract thought, tool, and arguments."""
269
+ thought = "No reasoning provided"
270
+ tool_name = "play_action"
271
+ tool_args = {"action": "look"}
272
+
273
+ lines = response.strip().split("\n")
274
 
275
+ for line in lines:
276
+ line_clean = line.strip()
277
+ line_upper = line_clean.upper()
278
+
279
+ if line_upper.startswith("THOUGHT:"):
280
+ thought = line_clean.split(":", 1)[1].strip()
281
+
282
+ elif line_upper.startswith("TOOL:"):
283
+ raw_tool = line_clean.split(":", 1)[1].strip().lower()
284
+ raw_tool = raw_tool.replace("**", "").replace("*", "").replace("`", "")
285
+ raw_tool = raw_tool.split()[0] if raw_tool else "play_action"
286
+ tool_name = raw_tool
287
+
288
+ elif line_upper.startswith("ARGS:"):
289
+ args_part = line_clean.split(":", 1)[1].strip()
290
+ try:
291
+ args_part = args_part.replace("'", '"')
292
+ tool_args = json.loads(args_part)
293
+ except json.JSONDecodeError:
294
+ match = re.search(r'"action"\s*:\s*"([^"]+)"', args_part)
295
+ if match:
296
+ tool_args = {"action": match.group(1)}
297
+ else:
298
+ tool_args = {"action": "look"}
299
+
300
+ return thought, tool_name, tool_args
301
 
302
+ def _validate_tool_call(self, tool_name: str, tool_args: dict, valid_tools: list[str]) -> tuple[str, dict]:
303
+ """Validate and fix common tool call issues."""
304
+ # Fix tool name
305
+ if tool_name not in valid_tools:
306
+ if tool_name in ["action", "do", "command"]:
307
+ tool_name = "play_action"
308
+ elif tool_name in ["map", "location"]:
309
+ tool_name = "get_map"
310
+ elif tool_name in ["mem", "state", "status"]:
311
+ tool_name = "memory"
312
+ elif tool_name in ["inv", "items"]:
313
+ tool_name = "inventory"
314
+ else:
315
+ tool_name = "play_action"
316
 
317
+ # Fix action verbs
318
+ if tool_name == "play_action":
319
+ action = tool_args.get("action", "look")
320
+
321
+ invalid_verb_map = {
322
+ "check": "examine",
323
+ "inspect": "examine",
324
+ "search": "look",
325
+ "grab": "take",
326
+ "pick": "take",
327
+ "use": "examine",
328
+ "investigate": "examine",
329
+ }
330
+
331
+ words = action.lower().split()
332
+ if words and words[0] in invalid_verb_map:
333
+ words[0] = invalid_verb_map[words[0]]
334
+ action = " ".join(words)
335
+
336
+ action = action.lower().strip()
337
+ action = action.replace("**", "").replace("*", "").replace("`", "")
338
+ action = " ".join(action.split())
339
+
340
+ tool_args["action"] = action
341
 
342
+ return tool_name, tool_args
343
+
344
+ def _extract_result(self, result) -> str:
345
+ """Extract text from MCP tool result."""
346
+ if hasattr(result, 'content') and result.content:
347
+ return result.content[0].text
348
+ if isinstance(result, list) and result:
349
+ return result[0].text if hasattr(result[0], 'text') else str(result[0])
350
+ return str(result)
351
 
352
+ def _update_score(self, text: str) -> None:
353
+ """Update score from game text."""
354
+ patterns = [
355
+ r'Score:\s*(\d+)',
356
+ r'score[:\s]+(\d+)',
357
+ r'\[Score:\s*(\d+)',
358
+ ]
359
 
360
+ for pattern in patterns:
361
+ match = re.search(pattern, text, re.IGNORECASE)
362
+ if match:
363
+ self.score = max(self.score, int(match.group(1)))
364
+
365
+ def _is_game_over(self, text: str) -> bool:
366
+ """Check if the game is over."""
367
+ game_over_phrases = [
368
+ "game over",
369
+ "you have died",
370
+ "you are dead",
371
+ "*** you have died ***",
372
+ ]
373
+ text_lower = text.lower()
374
+ return any(phrase in text_lower for phrase in game_over_phrases)
375
 
376
 
377
  # =============================================================================
378
+ # Local Testing
379
  # =============================================================================
380
 
381
  async def test_agent():
382
  """Test the agent locally."""
383
  from fastmcp import Client
384
 
 
 
 
385
  agent = StudentAgent()
386
+
387
+ #########
388
 
389
+ async with Client("mcp_server.py") as client:
390
  result = await agent.run(
391
  client=client,
392
  game="zork1",
393
+ max_steps=20,
394
  seed=42,
395
  verbose=True,
396
  )
397
 
398
+ print(f"\n{'=' * 50}")
399
+ print(f"Final Score: {result.final_score}")
400
  print(f"Moves: {result.moves}")
401
+ print(f"Locations: {len(result.locations_visited)}")
402
 
403
 
404
  if __name__ == "__main__":
mcp_server.py CHANGED
@@ -1,27 +1,8 @@
1
  """
2
- Student MCP Server for Text Adventure Games
3
 
4
- This is your MCP server submission. Implement the tools that your agent
5
- will use to play text adventure games.
6
-
7
- Required tool:
8
- play_action(action: str) -> str
9
- Execute a game command and return the result.
10
-
11
- Recommended tools:
12
- memory() -> str
13
- Return current game state, score, and recent history.
14
-
15
- inventory() -> str
16
- Return the player's current inventory.
17
-
18
- get_map() -> str
19
- Return a map of explored locations.
20
-
21
- Test your server with:
22
- fastmcp dev submission_template/mcp_server.py
23
-
24
- Then open the MCP Inspector in your browser to test the tools interactively.
25
  """
26
 
27
  import sys
@@ -31,179 +12,185 @@ import os
31
  sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
32
 
33
  from fastmcp import FastMCP
34
- from games.zork_env import TextAdventureEnv
35
 
36
 
37
- # =============================================================================
38
- # Create the MCP Server
39
- # =============================================================================
40
-
41
- mcp = FastMCP("Student Text Adventure Server")
42
 
 
 
43
 
44
- # =============================================================================
45
- # Game State Management
46
- # =============================================================================
47
 
48
- class GameManager:
49
- """
50
- Manages the text adventure game state.
51
 
52
- TODO: Extend this class to track:
53
- - Action history (for memory tool)
54
- - Explored locations (for mapping)
55
- - Current score and moves
56
- """
57
-
58
- def __init__(self):
59
- self.env: TextAdventureEnv = None
60
- self.state = None
61
- self.game_name: str = ""
62
- # TODO: Add more state tracking
63
- # self.history: list[tuple[str, str]] = []
64
- # self.explored_locations: dict[str, set[str]] = {}
65
- # self.current_location: str = ""
66
-
67
- def initialize(self, game: str = "zork1"):
68
- """Initialize or reset the game."""
69
  self.game_name = game
70
  self.env = TextAdventureEnv(game)
71
  self.state = self.env.reset()
72
- # TODO: Reset your state tracking here
73
- return self.state.observation
 
74
 
75
- def step(self, action: str) -> str:
76
- """Execute an action and return the result."""
77
- if self.env is None:
78
- self.initialize()
79
-
 
 
80
  self.state = self.env.step(action)
 
 
 
 
 
 
81
 
82
- # TODO: Update your state tracking here
83
- # self.history.append((action, self.state.observation))
84
- # Update location tracking, etc.
 
 
 
 
 
 
85
 
86
- return self.state.observation
87
 
88
- def get_score(self) -> int:
89
- """Get current score."""
90
- return self.state.score if self.state else 0
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
91
 
92
- def get_moves(self) -> int:
93
- """Get number of moves taken."""
94
- return self.state.moves if self.state else 0
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
95
 
96
 
97
- # Global game manager
98
- _game = GameManager()
99
 
100
 
101
- def get_game() -> GameManager:
102
- """Get or initialize the game manager."""
103
- global _game
104
- if _game.env is None:
105
- # Get game from environment variable (set by evaluator)
106
- game = os.environ.get("GAME", "zork1")
107
- _game.initialize(game)
108
- return _game
109
 
110
 
111
  # =============================================================================
112
- # MCP Tools - IMPLEMENT THESE
113
  # =============================================================================
114
 
115
  @mcp.tool()
116
  def play_action(action: str) -> str:
117
  """
118
- Execute a game command and return the result.
119
-
120
- This is the main tool for interacting with the game.
121
 
122
  Args:
123
- action: The command to execute (e.g., "north", "take lamp", "open mailbox")
124
-
125
  Returns:
126
- The game's response to the action
127
-
128
- Valid commands include:
129
- - Movement: north, south, east, west, up, down, enter, exit
130
- - Objects: take <item>, drop <item>, open <thing>, examine <thing>
131
- - Other: look, inventory, read <thing>, turn on lamp
132
  """
133
  game = get_game()
 
 
 
 
 
 
 
134
 
135
- # TODO: You might want to add action validation here
136
- # TODO: You might want to include score changes in the response
 
137
 
138
- result = game.step(action)
 
 
 
 
 
 
139
 
140
- # Optional: Append score info
141
- # result += f"\n[Score: {game.get_score()} | Moves: {game.get_moves()}]"
 
 
 
 
 
 
 
142
 
143
- return result
144
-
145
-
146
- # TODO: Implement additional tools to help your agent
147
-
148
- # @mcp.tool()
149
- # def memory() -> str:
150
- # """
151
- # Get the current game state summary.
152
- #
153
- # Returns:
154
- # A summary including current location, score, moves, and recent history
155
- # """
156
- # game = get_game()
157
- # # TODO: Return useful state information
158
- # pass
159
-
160
-
161
- # @mcp.tool()
162
- # def inventory() -> str:
163
- # """
164
- # Check what the player is carrying.
165
- #
166
- # Returns:
167
- # List of items in the player's inventory
168
- # """
169
- # game = get_game()
170
- # result = game.step("inventory")
171
- # return result
172
-
173
-
174
- # @mcp.tool()
175
- # def get_map() -> str:
176
- # """
177
- # Get a map of explored locations.
178
- #
179
- # Returns:
180
- # A text representation of explored locations and connections
181
- # """
182
- # game = get_game()
183
- # # TODO: Return map of explored locations
184
- # pass
185
-
186
-
187
- # @mcp.tool()
188
- # def get_valid_actions() -> str:
189
- # """
190
- # Get a list of likely valid actions from the current location.
191
- #
192
- # Returns:
193
- # List of actions that might work here
194
- # """
195
- # # This is a hint: Jericho provides get_valid_actions()
196
- # game = get_game()
197
- # if game.env and game.env.env:
198
- # valid = game.env.env.get_valid_actions()
199
- # return "Valid actions: " + ", ".join(valid[:20])
200
- # return "Could not determine valid actions"
201
 
202
 
203
  # =============================================================================
204
- # Run the server
205
  # =============================================================================
206
 
207
  if __name__ == "__main__":
208
- # This runs the server with stdio transport (for MCP clients)
209
- mcp.run()
 
1
  """
2
+ Example: MCP Server for Text Adventures
3
 
4
+ A complete MCP server that exposes text adventure games via tools.
5
+ This demonstrates a full-featured server with memory, mapping, and inventory.
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
6
  """
7
 
8
  import sys
 
12
  sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
13
 
14
  from fastmcp import FastMCP
15
+ from games.zork_env import TextAdventureEnv, list_available_games
16
 
17
 
18
+ # Get game from environment variable (default: zork1)
19
+ INITIAL_GAME = os.environ.get("GAME", "zork1")
 
 
 
20
 
21
+ # Create the MCP server
22
+ mcp = FastMCP("Text Adventure Server")
23
 
 
 
 
24
 
25
+ class GameState:
26
+ """Manages the text adventure game state and exploration data."""
 
27
 
28
+ def __init__(self, game: str = "zork1"):
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
29
  self.game_name = game
30
  self.env = TextAdventureEnv(game)
31
  self.state = self.env.reset()
32
+ self.history: list[tuple[str, str]] = []
33
+ self.explored_locations: dict[str, set[str]] = {}
34
+ self.current_location: str = self._extract_location(self.state.observation)
35
 
36
+ def _extract_location(self, observation: str) -> str:
37
+ """Extract location name from observation (usually first line)."""
38
+ lines = observation.strip().split('\n')
39
+ return lines[0] if lines else "Unknown"
40
+
41
+ def take_action(self, action: str) -> str:
42
+ """Execute a game action and return the result."""
43
  self.state = self.env.step(action)
44
+ result = self.state.observation
45
+
46
+ # Track history
47
+ self.history.append((action, result))
48
+ if len(self.history) > 50:
49
+ self.history = self.history[-50:]
50
 
51
+ # Update map
52
+ new_location = self._extract_location(result)
53
+ if action in ["north", "south", "east", "west", "up", "down",
54
+ "enter", "exit", "n", "s", "e", "w", "u", "d"]:
55
+ if self.current_location not in self.explored_locations:
56
+ self.explored_locations[self.current_location] = set()
57
+ if new_location != self.current_location:
58
+ self.explored_locations[self.current_location].add(f"{action} -> {new_location}")
59
+ self.current_location = new_location
60
 
61
+ return result
62
 
63
+ def get_memory(self) -> str:
64
+ """Get a summary of current game state."""
65
+ recent = self.history[-5:] if self.history else []
66
+ recent_str = "\n".join([f" > {a} -> {r[:60]}..." for a, r in recent]) if recent else " (none yet)"
67
+
68
+ return f"""Current State:
69
+ - Location: {self.current_location}
70
+ - Score: {self.state.score} points
71
+ - Moves: {self.state.moves}
72
+ - Game: {self.game_name}
73
+
74
+ Recent Actions:
75
+ {recent_str}
76
+
77
+ Current Observation:
78
+ {self.state.observation}"""
79
+
80
+ def get_map(self) -> str:
81
+ """Get a map of explored locations."""
82
+ if not self.explored_locations:
83
+ return "Map: No locations explored yet. Try moving around!"
84
+
85
+ lines = ["Explored Locations and Exits:"]
86
+ for loc, exits in sorted(self.explored_locations.items()):
87
+ lines.append(f"\n* {loc}")
88
+ for exit_info in sorted(exits):
89
+ lines.append(f" -> {exit_info}")
90
+
91
+ lines.append(f"\n[Current] {self.current_location}")
92
+ return "\n".join(lines)
93
 
94
+ def get_inventory(self) -> str:
95
+ """Get current inventory."""
96
+ items = self.state.inventory if hasattr(self.state, 'inventory') and self.state.inventory else []
97
+
98
+ if not items:
99
+ return "Inventory: You are empty-handed."
100
+
101
+ item_names = []
102
+ for item in items:
103
+ item_str = str(item)
104
+ item_lower = item_str.lower()
105
+ if "parent" in item_lower:
106
+ idx = item_lower.index("parent")
107
+ name = item_str[:idx].strip()
108
+ if ":" in name:
109
+ name = name.split(":", 1)[1].strip()
110
+ item_names.append(name)
111
+ elif ":" in item_str:
112
+ name = item_str.split(":")[1].strip()
113
+ item_names.append(name)
114
+ else:
115
+ item_names.append(item_str)
116
+
117
+ return f"Inventory: {', '.join(item_names)}"
118
 
119
 
120
+ # Global game state
121
+ _game_state: GameState | None = None
122
 
123
 
124
+ def get_game() -> GameState:
125
+ """Get or initialize the game state."""
126
+ global _game_state
127
+ if _game_state is None:
128
+ _game_state = GameState(INITIAL_GAME)
129
+ return _game_state
 
 
130
 
131
 
132
  # =============================================================================
133
+ # MCP Tools
134
  # =============================================================================
135
 
136
  @mcp.tool()
137
  def play_action(action: str) -> str:
138
  """
139
+ Execute a game action in the text adventure.
 
 
140
 
141
  Args:
142
+ action: The command to execute (e.g., 'north', 'take lamp', 'open mailbox')
143
+
144
  Returns:
145
+ The game's response to your action
 
 
 
 
 
146
  """
147
  game = get_game()
148
+ result = game.take_action(action)
149
+
150
+ # Add score info
151
+ score_info = f"\n\n[Score: {game.state.score} | Moves: {game.state.moves}]"
152
+
153
+ if game.state.reward > 0:
154
+ score_info = f"\n\n+{game.state.reward} points! (Total: {game.state.score})"
155
 
156
+ done_info = ""
157
+ if game.state.done:
158
+ done_info = "\n\nGAME OVER"
159
 
160
+ return result + score_info + done_info
161
+
162
+
163
+ @mcp.tool()
164
+ def memory() -> str:
165
+ """
166
+ Get a summary of the current game state.
167
 
168
+ Returns location, score, moves, recent actions, and current observation.
169
+ """
170
+ return get_game().get_memory()
171
+
172
+
173
+ @mcp.tool()
174
+ def get_map() -> str:
175
+ """
176
+ Get a map showing explored locations and connections.
177
 
178
+ Useful for navigation and avoiding getting lost.
179
+ """
180
+ return get_game().get_map()
181
+
182
+
183
+ @mcp.tool()
184
+ def inventory() -> str:
185
+ """
186
+ Check what items you are currently carrying.
187
+ """
188
+ return get_game().get_inventory()
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
189
 
190
 
191
  # =============================================================================
192
+ # Main
193
  # =============================================================================
194
 
195
  if __name__ == "__main__":
196
+ mcp.run()