create chess agent

README.md

@@ -1,6 +1,7 @@
 ---
+tags: [agents, agent-demo-track, chess, chessboard, games]
 title: Chess Agent
-emoji: 🏆
+emoji: ♟️
 colorFrom: indigo
 colorTo: purple
 sdk: gradio
@@ -10,5 +11,3 @@ pinned: false
 license: apache-2.0
 short_description: Chess Agent
 ---
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,152 @@
+import asyncio
+import os
+
+import chess
+import gradio as gr
+from gradio_chessboard import Chessboard
+from langchain_core.messages import HumanMessage
+from utils.helpers import call_agent, create_agent
+from utils.tools import create_base_tools, create_mcp_tools
+
+MODEL_DEFAULTS = {
+    "anthropic": {
+        "model_name": "claude-sonnet-4-20250514",
+        "provider": "anthropic",
+        "api_key": os.getenv("ANTHROPIC_API_KEY", ""),
+    },
+    "mistral": {
+        "model_name": "mistral-large-latest",
+        "provider": "mistralai",
+        "api_key": os.getenv("MISTRAL_API_KEY", ""),
+    },
+    "openai": {
+        "model_name": "gpt-4o",
+        "provider": "openai",
+        "api_key": os.getenv("OPENAI_API_KEY", ""),
+    },
+}
+
+
+async def main():
+    board = chess.Board()
+
+    base_tools = create_base_tools(board)
+    mcp_tools = await create_mcp_tools(
+        url="https://czakop-chess-agent-mcp.hf.space/gradio_api/mcp/sse",
+        transport="sse",
+    )
+
+    async def chat_entrypoint(prompt, messages, model_name, model_provider, api_key):
+        """Entrypoint for the chat interaction."""
+        messages.append(gr.ChatMessage(role="user", content=prompt))
+        yield messages, board.fen()
+
+        real_prompt = HumanMessage(
+            content=f"{prompt}\nCurrent board state: {board.fen()}"
+        )
+        agent = create_agent(
+            model_name, model_provider, api_key, base_tools + mcp_tools
+        )
+        async for messages in call_agent(agent, messages, real_prompt):
+            yield messages, board.fen()
+
+    async def move_entrypoint(messages, fen, model_name, model_provider, api_key):
+        """Entrypoint for the chess move interaction."""
+        board.set_fen(fen)
+        messages.append(
+            gr.ChatMessage(
+                role="user",
+                content="Your turn! Make a move.",
+            )
+        )
+        yield messages, board.fen()
+
+        real_prompt = HumanMessage(
+            content=f"Make a move with {'white' if board.turn == chess.WHITE else 'black'}, current board state: {fen}. Use can use tools to analyze the position."
+        )
+        agent = create_agent(
+            model_name, model_provider, api_key, base_tools + mcp_tools
+        )
+        async for messages in call_agent(agent, messages, real_prompt):
+            yield messages, board.fen()
+
+    with gr.Blocks(fill_height=True) as chessagent:
+        gr.Markdown("# Play Chess with an AI Agent ♔ and see its thoughts 💭")
+
+        with gr.Row():
+            with gr.Column(min_width=500):
+                with gr.Row():
+                    model_provider = gr.Dropdown(
+                        choices=["Anthropic", "Mistral", "OpenAI"],
+                        value="Anthropic",
+                        label="Model Provider",
+                        type="value",
+                        interactive=True,
+                        allow_custom_value=False,
+                    )
+                    model_name = gr.Textbox(
+                        value=MODEL_DEFAULTS["anthropic"]["model_name"],
+                        label="Model Name",
+                        placeholder="Enter model name",
+                        interactive=True,
+                    )
+                    api_key = gr.Textbox(
+                        value=MODEL_DEFAULTS["anthropic"]["api_key"],
+                        label="API Key",
+                        placeholder="Enter your API key",
+                        type="password",
+                        interactive=True,
+                    )
+                    model_provider.change(
+                        fn=lambda provider: MODEL_DEFAULTS[provider.lower()][
+                            "model_name"
+                        ],
+                        inputs=model_provider,
+                        outputs=model_name,
+                    )
+                    model_provider.change(
+                        fn=lambda provider: MODEL_DEFAULTS[provider.lower()]["api_key"],
+                        inputs=model_provider,
+                        outputs=api_key,
+                    )
+
+                board_component = Chessboard(game_mode=True, label="Chess Board")
+
+                chatbot = gr.Chatbot(
+                    type="messages",
+                    label="Chess Agent",
+                    avatar_images=(
+                        "https://chessboardjs.com/img/chesspieces/wikipedia/wK.png",
+                        "https://chessboardjs.com/img/chesspieces/wikipedia/bK.png",
+                    ),
+                    min_height=650,
+                    render=False,
+                )
+
+                board_component.move(
+                    fn=move_entrypoint,
+                    inputs=[
+                        chatbot,
+                        board_component,
+                        model_name,
+                        model_provider,
+                        api_key,
+                    ],
+                    outputs=[chatbot, board_component],
+                )
+
+            with gr.Column():
+                chatbot.render()
+                input_box = gr.Textbox(lines=1, label="Chat Message")
+                input_box.submit(
+                    fn=chat_entrypoint,
+                    inputs=[input_box, chatbot, model_name, model_provider, api_key],
+                    outputs=[chatbot, board_component],
+                )
+                input_box.submit(lambda: "", None, [input_box], queue=False)
+
+        chessagent.launch()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

requirements.txt

@@ -0,0 +1,9 @@
+gradio
+gradio_chessboard
+chess
+langchain
+langchain-mcp-adapters
+langchain-anthropic
+langchain-mistralai
+langchain-openai
+langgraph

utils/helpers.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, AsyncIterator
+
+import gradio as gr
+from langchain.chat_models import init_chat_model
+from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
+from langgraph.prebuilt import create_react_agent
+
+if TYPE_CHECKING:
+    from langchain_core.language_models.chat_models import BaseChatModel
+    from langgraph.graph.graph import CompiledGraph
+
+MESSAGE_TYPE = BaseMessage | gr.ChatMessage | dict[str, str]
+
+
+def create_agent(
+    model_name: str, provider: str, api_key: str, tools: list
+) -> CompiledGraph:
+    """Create a React agent with the specified model."""
+    model = _create_model(model_name, provider, api_key)
+    return create_react_agent(
+        model,
+        tools=tools,
+    )
+
+
+async def call_agent(
+    agent: CompiledGraph, messages: list[MESSAGE_TYPE], prompt: HumanMessage
+) -> AsyncIterator[list[MESSAGE_TYPE]]:
+    async for chunk in agent.astream(
+        {
+            "messages": [_convert_to_langchain_message(msg) for msg in messages[:-1]]
+            + [prompt]
+        }
+    ):
+        print(f"Chunk received: {chunk}")
+        if "tools" in chunk:
+            for step in chunk["tools"]["messages"]:
+                messages.append(
+                    gr.ChatMessage(
+                        role="assistant",
+                        content=step.content,
+                        metadata={"title": f"🛠️ Used tool {step.name}"},
+                    )
+                )
+                yield messages
+        if "agent" in chunk:
+            messages.append(
+                gr.ChatMessage(
+                    role="assistant",
+                    content=_get_chunk_message_content(chunk),
+                )
+            )
+            yield messages
+
+
+def _create_model(model_name: str, provider: str, api_key: str) -> BaseChatModel:
+    """Get the chat model based on the provider and model name."""
+    if provider == "Anthropic":
+        return init_chat_model(
+            "anthropic:" + model_name,
+            anthropic_api_key=api_key,
+        )
+    elif provider == "Mistral":
+        return init_chat_model(
+            "mistralai:" + model_name,
+            mistral_api_key=api_key,
+        )
+    elif provider == "OpenAI":
+        return init_chat_model(
+            "openai:" + model_name,
+            openai_api_key=api_key,
+        )
+    else:
+        raise ValueError(f"Unsupported model provider: {provider}")
+
+
+def _is_ai_message(message: MESSAGE_TYPE) -> bool:
+    if isinstance(message, AIMessage):
+        return True
+    if isinstance(message, gr.ChatMessage):
+        return message.role == "assistant"
+    if isinstance(message, dict):
+        return message.get("role") == "assistant"
+    return False
+
+
+def _convert_to_langchain_message(message: MESSAGE_TYPE) -> BaseMessage:
+    if isinstance(message, BaseMessage):
+        return message
+    if isinstance(message, gr.ChatMessage):
+        return (
+            AIMessage(content=message.content)
+            if _is_ai_message(message)
+            else HumanMessage(content=message.content)
+        )
+    if isinstance(message, dict):
+        return (
+            AIMessage(content=message.get("content", ""))
+            if _is_ai_message(message)
+            else HumanMessage(content=message.get("content", ""))
+        )
+    raise ValueError(f"Unsupported message type: {type(message)}")
+
+
+def _get_chunk_message_content(chunk: dict) -> str:
+    msg_object = chunk["agent"]["messages"][0]
+    message = msg_object.content
+    if isinstance(message, list):
+        message = message[0] if message else ""
+    if isinstance(message, dict):
+        message = message.get("text")
+    return message or "Calling tool(s)"

utils/tools.py

@@ -0,0 +1,71 @@
+import chess
+from langchain_core.tools import BaseTool, tool
+from langchain_mcp_adapters.client import MultiServerMCPClient
+
+
+def create_base_tools(board: chess.Board) -> list[BaseTool]:
+    """Create tools for interacting with a chess board.
+
+    Args:
+        board (chess.Board): The chess board to interact with.
+
+    Returns:
+        list[BaseTool]: A list of tools for interacting with the chess board.
+    """
+
+    @tool
+    def get_fen() -> str:
+        """Get the current FEN string of the chess board."""
+        return board.fen()
+
+    @tool
+    def set_fen(fen: str) -> str:
+        """Set the chess board to a specific FEN string.
+        Don't use when you are playing a game, use the `make_move` tool instead.
+
+        Args:
+            fen (str): The FEN string to set the board to.
+        """
+        try:
+            board.set_fen(fen)
+            return board.fen()
+        except ValueError as e:
+            return str(e)
+
+    @tool
+    def make_move(move: str) -> str:
+        """Make a move on the chess board and return the new FEN string.
+
+        Args:
+            move (str): The move in UCI format (e.g., "e2e4").
+        """
+        try:
+            chess_move = chess.Move.from_uci(move)
+            if chess_move in board.legal_moves:
+                board.push(chess_move)
+                return board.fen()
+            else:
+                return "Illegal move"
+        except Exception as e:
+            return str(e)
+
+    return [
+        get_fen,
+        set_fen,
+        make_move,
+    ]
+
+
+async def create_mcp_tools(
+    url: str = "http://localhost:7860/gradio_api/mcp/sse", transport: str = "sse"
+) -> list[BaseTool]:
+    mcp_client = MultiServerMCPClient(
+        {
+            "chess": {
+                "url": url,
+                "transport": transport,
+            }
+        }
+    )
+
+    return await mcp_client.get_tools()