212 lines
6.7 KiB
Python
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()
|