mcp_servers/boss_server.py

import logging
import os

from langchain_core.messages import ToolMessage
from langgraph.types import Command
from mcp.server.fastmcp import FastMCP
from support.tools.general import convert_to_string
from typing import Union, Any, List, Dict
from uuid import uuid4


os.makedirs("logs", exist_ok=True)
logging.basicConfig(
    filename='logs/boss_server.log',                    # Log file name
    filemode='a',                          # 'a' for append, 'w' to overwrite
    level=logging.INFO,                    # Minimum level to log
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger(__name__)


mcp = FastMCP("BossServer")


@mcp.tool()
async def ChooseWord(
    clue_number: int,
    clue: Union[str, Dict, List, Any],
):
    """Use this tool if you think that some of the informations are missing.
    Args:
        clue_number: the number of word of your team that can be linked to the word.
        clue: a single word that can be linked to {clue_number} words on the board."""

    clue = convert_to_string(clue)

    logger.info(f"Chosen clue: {clue} for number: {clue_number}")

    return Command(
        # goto=f"{team_color}_captain",
        update={
            "clue_number": clue_number,
            "clue": clue,
            "messages": [ToolMessage(
                content=f"{clue}, {clue_number}",
                name="ChooseWord",
                tool_call_id=str(uuid4())
            )],
        }
    )


def start_boss_server():
    mcp.settings.port = 8000
    mcp.run(transport="stdio")


if __name__ == "__main__":
    start_boss_server()