File size: 2,503 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 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 |
import logging
import os
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from mcp.server.fastmcp import FastMCP
from typing import Union, Any, List, Dict
from uuid import uuid4
os.makedirs("logs", exist_ok=True)
logging.basicConfig(
filename='logs/captain_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("CaptainServer")
@mcp.tool()
async def TeamFinalChoice(
guesses: List[str],
):
"""Use this tool if you think that some of the informations are missing.
Args:
guesses: a list of words (string) that you think are associated to the word told by the Boss Agent."""
logger.info(f"Final choices made: {guesses}")
artifact = {"guesses": guesses}
return Command(
update={
"guesses": guesses,
"messages": [ToolMessage(
content="I made my final choices: " + ", ".join(guesses),
name="TeamFinalChoice",
artifact=artifact,
tool_call_id=str(uuid4())
)],
}
)
@mcp.tool()
async def Call_Agent_1(
message: Union[str, Dict, List, Any],
):
"""Use this tool to call Agent 1 for help.
Args:
message: a message for Agent 1."""
logger.info(f"Calling Agent 1 with message: {message}")
return Command(
update={
"round_messages": [message],
"messages": [ToolMessage(
content=message,
name="Call_Agent_1",
tool_call_id=str(uuid4())
)],
}
)
@mcp.tool()
async def Call_Agent_2(
message: Union[str, Dict, List, Any],
):
"""Use this tool to call Agent 2 for help.
Args:
message: a message for Agent 2."""
logger.info(f"Calling Agent 2 with message: {message}")
return Command(
update={
"round_messages": [message],
"messages": [ToolMessage(
content=message,
name="Call_Agent_2",
tool_call_id=str(uuid4())
)],
}
)
def start_captain_server():
mcp.settings.port = 8001
mcp.run(transport="stdio")
if __name__ == "__main__":
start_captain_server()
|