elo-tac-toe-mcp/agent.py

212 lines
6.7 KiB
Python

#!/usr/bin/env python3
"""
Simple Python agent that plays ELO Tic-Tac-Toe via the MCP tools.
This demonstrates how to integrate with the game using MCP.
"""
import json
import subprocess
import sys
import time
import uuid
def call_tool(tool_name: str, **kwargs) -> dict:
"""Call an MCP tool and return the parsed result."""
# For demonstration, we'll use direct HTTP calls instead of MCP
# In a real MCP setup, you'd use the MCP client library
import urllib.request
import urllib.parse
base_url = "http://localhost:8080" # or your server URL
api_key = getattr(call_tool, '_api_key', None)
if not api_key:
# Register a new agent
req = urllib.request.Request(
f"{base_url}/auth/register",
data=json.dumps({"name": f"python-agent-{uuid.uuid4().hex[:8]}"}).encode(),
headers={"Content-Type": "application/json"},
method="POST"
)
resp = urllib.request.urlopen(req)
data = json.loads(resp.read().decode())
call_tool._api_key = data["apiKey"]
call_tool._agent_id = data["agentId"]
print(f"Registered agent: {data['agentId']}")
api_key = call_tool._api_key
# Get session token
req = urllib.request.Request(
f"{base_url}/auth/session",
data=json.dumps({"apiKey": api_key}).encode(),
headers={"Content-Type": "application/json"},
method="POST"
)
resp = urllib.request.urlopen(req)
data = json.loads(resp.read().decode())
token = data["token"]
# Map tool names to API endpoints
endpoints = {
"elo_tac_toe_join_queue": ("POST", "/queue/join", lambda args: {"gameType": "tictactoe", "mode": args.get("mode", "ranked")}),
"elo_tac_toe_leave_queue": ("POST", "/queue/leave", lambda args: {"gameType": "tictactoe", "mode": args.get("mode", "ranked")}),
"elo_tac_toe_wait_match": ("GET", "/match/next", lambda args: {}),
"elo_tac_toe_get_turn_state": ("GET", f"/game/{{gameId}}/state", lambda args: {}),
"elo_tac_toe_submit_move": ("POST", f"/game/{{gameId}}/move", lambda args: {"cell": args["cell"], "idempotencyKey": args.get("idempotencyKey", str(uuid.uuid4()))}),
"elo_tac_toe_resign": ("POST", f"/game/{{gameId}}/resign", lambda args: {}),
"elo_tac_toe_my_rating": ("GET", "/agent/me/rating", lambda args: {}),
}
if tool_name not in endpoints:
raise ValueError(f"Unknown tool: {tool_name}")
method, path, body_fn = endpoints[tool_name]
# Format path with gameId if needed
if "gameId" in kwargs:
path = path.format(gameId=kwa rgs["gameId"])
body = body_fn(kwargs)
req = urllib.request.Request(
f"{base_url}{path}",
data=json.dumps(body).encode() if body and method == "POST" else None,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {token}",
},
method=method
)
resp = urllib.request.urlopen(req, timeout=60)
return json.loads(resp.read().decode())
def calculate_best_move(board: list, my_mark: str) -> int:
"""
Simple tic-tac-toe AI:
1. Win if possible
2. Block opponent if they can win
3. Take center
4. Take corners
5. Take random available
"""
opponent = 'o' if my_mark == 'x' else 'x'
wins = [(0,1,2), (3,4,5), (6,7,8), (0,3,6), (1,4,7), (2,5,8), (0,4,8), (2,4,6)]
def check_win(positions, mark):
return all(board[p] == mark for p in positions)
# 1. Win if possible
for move in range(9):
if board[move] is None:
board[move] = my_mark
if any(check_win(w, my_mark) for w in wins):
board[move] = None
return move + 1 # 1-indexed
board[move] = None
# 2. Block opponent
for move in range(9):
if board[move] is None:
board[move] = opponent
if any(check_win(w, opponent) for w in wins):
board[move] = None
return move + 1
board[move] = None
# 3. Take center
if board[4] is None:
return 5
# 4. Take corners
for corner in [0, 2, 6, 8]:
if board[corner] is None:
return corner + 1
# 5. Any available
for move in range(9):
if board[move] is None:
return move + 1
return 5 # Default
def play_game():
"""Play a complete game of tic-tac-toe."""
print("\n=== Starting ELO Tic-Tac-Toe Game ===\n")
# Join queue
print("Joining ranked queue...")
result = call_tool("elo_tac_toe_join_queue", mode="ranked")
print(f"Queue status: {result}")
# Wait for match
print("Waiting for opponent...")
result = call_tool("elo_tac_toe_wait_match", timeoutMs=60000)
print(f"Match result: {result}")
if "gameId" not in result:
print("No match found!")
return
game_id = result["gameId"]
print(f"Matched! Game ID: {game_id}")
# Get game state
state = call_tool("elo_tac_toe_get_turn_state", gameId=game_id)
my_mark = state.get("yourMark", "x")
print(f"You are: {my_mark.upper()}")
# Play game
turn = 0
while turn < 10:
turn += 1
# Get current state
state = call_tool("elo_tac_toe_get_turn_state", gameId=game_id)
print(f"\n--- Turn {turn} ---")
print(f"Board: {state.get('board', [])}")
print(f"Status: {state.get('status', 'unknown')}")
# Check if game over
if state.get("status") == "game_over":
print(f"\nGame Over!")
print(f"Result: {state.get('result', {})}")
break
# Make move if it's our turn
if state.get("status") == "your_turn":
legal_moves = state.get("legalMoves", [])
if not legal_moves:
print("No legal moves!")
break
move = calculate_best_move(state.get("board", []), my_mark)
print(f"Making move: {move}")
result = call_tool(
"elo_tac_toe_submit_move",
gameId=game_id,
cell=move,
idempotencyKey=f"move-{turn}-{uuid.uuid4().hex[:8]}"
)
print(f"Move result: {result}")
time.sleep(0.5) # Small delay
# Get final rating
rating = call_tool("elo_tac_toe_my_rating")
print(f"\nFinal Rating: {rating.get('rating', 'N/A')}")
print(f"Games Played: {rating.get('gamesPlayed', 'N/A')}")
if __name__ == "__main__":
try:
play_game()
except Exception as e:
print(f"Error: {e}")
import traceback
traceback.print_exc()