File size: 1,651 Bytes
aa2d45f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 |
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()
|