chess_game/chess_server.py

import os
import random
from typing import Dict, List, Optional

import chess
import chess.pgn
from mcp.server.fastmcp import FastMCP


mcp = FastMCP(name="ChessServer", stateless_http=True)


# Simple in-memory game state keyed by session_id
games: Dict[str, Dict] = {}


def _get_session_id() -> str:
    # Provide a default single-session id when not given by client
    return "default"


def _ensure_game(session_id: str) -> Dict:
    if session_id not in games:
        games[session_id] = {
            "board": chess.Board(),
            "pgn_game": chess.pgn.Game(),
            "node": None,
        }
        games[session_id]["node"] = games[session_id]["pgn_game"]
    return games[session_id]


@mcp.tool(description="Start a new chess game. Returns initial board state.")
def start_game(session_id: Optional[str] = None, player_color: str = "white") -> Dict:
    sid = session_id or _get_session_id()
    game = {
        "board": chess.Board(),
        "pgn_game": chess.pgn.Game(),
        "node": None,
    }
    if player_color.lower() not in ("white", "black"):
        player_color = "white"
    # If AI should play first (player is black), make an AI opening move
    if player_color.lower() == "black":
        # Let engine choose a random legal move
        ai_move = random.choice(list(game["board"].legal_moves))
        game["board"].push(ai_move)
    game["node"] = game["pgn_game"]
    games[sid] = game
    return _board_state(game["board"]) | {"session_id": sid, "player_color": player_color}


def _board_state(board: chess.Board) -> Dict:
    return {
        "fen": board.fen(),
        "unicode": board.unicode(borders=True),
        "turn": "white" if board.turn else "black",
        "is_game_over": board.is_game_over(),
        "result": board.result() if board.is_game_over() else None,
        "legal_moves": [board.san(m) for m in board.legal_moves],
    }


@mcp.tool(description="Make a player move in SAN or UCI notation.")
def player_move(move: str, session_id: Optional[str] = None) -> Dict:
    sid = session_id or _get_session_id()
    g = _ensure_game(sid)
    board: chess.Board = g["board"]
    try:
        try:
            chess_move = board.parse_san(move)
        except ValueError:
            chess_move = chess.Move.from_uci(move)
            if chess_move not in board.legal_moves:
                raise ValueError("Illegal move")
        board.push(chess_move)
        # Update PGN
        g["node"] = g["node"].add_variation(chess_move)
        return _board_state(board) | {"last_move": board.san(chess_move)}
    except Exception as e:
        return {"error": f"Invalid move: {e}"}


@mcp.tool(description="Have the AI make a move (random legal move).")
def ai_move(session_id: Optional[str] = None) -> Dict:
    sid = session_id or _get_session_id()
    g = _ensure_game(sid)
    board: chess.Board = g["board"]
    if board.is_game_over():
        return _board_state(board)
    move = random.choice(list(board.legal_moves))
    board.push(move)
    g["node"] = g["node"].add_variation(move)
    return _board_state(board) | {"last_move": board.san(move)}


@mcp.tool(description="Return current board state.")
def board(session_id: Optional[str] = None) -> Dict:
    sid = session_id or _get_session_id()
    g = _ensure_game(sid)
    return _board_state(g["board"]) | {"session_id": sid}


@mcp.tool(description="List legal moves in SAN notation.")
def legal_moves(session_id: Optional[str] = None) -> List[str]:
    sid = session_id or _get_session_id()
    g = _ensure_game(sid)
    b: chess.Board = g["board"]
    return [b.san(m) for m in b.legal_moves]


@mcp.tool(description="Game status including check, checkmate, stalemate.")
def status(session_id: Optional[str] = None) -> Dict:
    sid = session_id or _get_session_id()
    g = _ensure_game(sid)
    b: chess.Board = g["board"]
    return {
        "turn": "white" if b.turn else "black",
        "is_check": b.is_check(),
        "is_checkmate": b.is_checkmate(),
        "is_stalemate": b.is_stalemate(),
        "is_insufficient_material": b.is_insufficient_material(),
        "is_game_over": b.is_game_over(),
        "result": b.result() if b.is_game_over() else None,
    }


@mcp.tool(description="Export game PGN.")
def pgn(session_id: Optional[str] = None) -> str:
    sid = session_id or _get_session_id()
    g = _ensure_game(sid)
    game = g["pgn_game"]
    exporter = chess.pgn.StringExporter(headers=True, variations=False, comments=False)
    return game.accept(exporter)