MCP Chat

Instructions

Implementation Reference

• mcp_chat/server.py:325-428 (handler)

  The core handler function for the 'wait_for_message' tool. It sets up an asyncio.Queue for the user in the room, waits for incoming messages with a configurable timeout (max 300s), handles timeout, cancellation, and errors, and cleans up the queue afterward. Messages are received via send_message calls from partners.
  @mcp.tool() async def wait_for_message( room_id: str, client_id: str, timeout: int = 60 ) -> Dict[str, Any]: """Wait for a message in the chat room (long-polling). This tool blocks until a message is received or the timeout is reached. Use this after sending a message to wait for a response, or call it first to wait for an incoming message. Conversation flow: - If you sent the last message: wait_for_message to get response - If you're waiting for first contact: wait_for_message before sending - After receiving a message: send_message to respond, then wait_for_message again Args: room_id: The ID of the chat room to listen in client_id: Your client identifier (from enter_queue or join_room) timeout: Timeout in seconds (default: 60, max: 300) Returns: On message: {"message": "text", "sender": "name", "timestamp": "...", "message_id": "..."} On timeout: {"timeout": true, "message": "No message received"} On error: {"error": "error message"} """ # Use the provided client_id connection_id = client_id # Validate timeout timeout = min(timeout, 300) # Max 5 minutes timeout = max(timeout, 1) # Min 1 second # Get user user = connections.get(connection_id) if not user: logger.error(f"User not found for client_id: {client_id}") logger.debug(f"Active connections: {list(connections.keys())}") return {"error": f"User not found. Invalid client_id: {client_id}"} # Get and validate room room = await room_manager.get_room(room_id) if not room: return {"error": "Room not found"} if not room.active: return {"error": "Chat has ended"} # Verify user is in the room if not room.has_user(user.user_id): return {"error": "You are not in this room"} # Create a message queue for this user if not exists if room_id not in message_queues: message_queues[room_id] = {} # Create queue with reasonable size limit message_queue: asyncio.Queue[Dict[str, Any]] = asyncio.Queue(maxsize=100) message_queues[room_id][user.user_id] = message_queue logger.info( f"User {user.name} waiting for messages in room {room_id} (timeout: {timeout}s)" ) try: # Wait for a message with timeout message_data = await asyncio.wait_for( message_queue.get(), timeout=float(timeout) ) logger.info( f"Message received for {user.name}: {message_data.get('content', '')[:50]}..." ) return { "message": message_data["content"], "sender": message_data["sender_name"], "timestamp": message_data["timestamp"], "message_id": message_data["message_id"], } except asyncio.TimeoutError: logger.info(f"Timeout waiting for message for {user.name}") return {"timeout": True, "message": "No message received within timeout period"} except asyncio.CancelledError: # Client cancelled the request - this is normal behavior logger.info(f"Wait cancelled for {user.name}") # Re-raise to let the framework handle it properly raise except Exception as e: logger.error(f"Error in wait_for_message: {e}") return {"error": f"Unexpected error: {str(e)}"} finally: # Clean up queue registration if room_id in message_queues and user.user_id in message_queues[room_id]: del message_queues[room_id][user.user_id] # Clean up empty room entries if not message_queues[room_id]: del message_queues[room_id] logger.info(f"Cleaned up message queue for {user.name}")
• mcp_chat/server.py:325-325 (registration)

  Registration of the 'wait_for_message' tool using the FastMCP @mcp.tool() decorator, which automatically generates schema from function signature and docstring.
  @mcp.tool()