From 53f672bb7697b3d2c5a6d0e5e3a9523ba5584d25 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Claude4=CE=9Elope?= Date: Fri, 27 Jun 2025 23:50:45 +0200 Subject: [PATCH] Add multiuser support to screen server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add enable_multiuser tool to enable multiuser mode on sessions - Add attach_multiuser tool to provide attach instructions - Add add_user tool for access control list management - Peek functionality works seamlessly with multiuser sessions - Create comprehensive test for multiuser functionality - All existing functionality preserved and tested 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- mcp_servers/screen/screen_server.py | 107 +++++++++++++++++ test_screen_multiuser.py | 172 ++++++++++++++++++++++++++++ 2 files changed, 279 insertions(+) create mode 100644 test_screen_multiuser.py diff --git a/mcp_servers/screen/screen_server.py b/mcp_servers/screen/screen_server.py index a6202e9..125d2f4 100755 --- a/mcp_servers/screen/screen_server.py +++ b/mcp_servers/screen/screen_server.py @@ -115,6 +115,62 @@ class ScreenServer(BaseMCPServer): "required": ["session"] } ) + + # Enable multiuser mode + self.register_tool( + name="enable_multiuser", + description="Enable multiuser mode for a screen session", + input_schema={ + "type": "object", + "properties": { + "session": { + "type": "string", + "description": "Name of the screen session" + } + }, + "required": ["session"] + } + ) + + # Attach to multiuser session + self.register_tool( + name="attach_multiuser", + description="Attach to a multiuser screen session (for external use)", + input_schema={ + "type": "object", + "properties": { + "session": { + "type": "string", + "description": "Name of the screen session" + }, + "user": { + "type": "string", + "description": "Optional username for access control" + } + }, + "required": ["session"] + } + ) + + # Add user to multiuser session + self.register_tool( + name="add_user", + description="Add a user to a multiuser screen session", + input_schema={ + "type": "object", + "properties": { + "session": { + "type": "string", + "description": "Name of the screen session" + }, + "user": { + "type": "string", + "description": "Username to add" + } + }, + "required": ["session", "user"] + } + ) async def handle_tool_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """Handle screen tool calls.""" @@ -129,6 +185,12 @@ class ScreenServer(BaseMCPServer): return await self._list_sessions() elif tool_name == "kill_session": return await self._kill_session(arguments) + elif tool_name == "enable_multiuser": + return await self._enable_multiuser(arguments) + elif tool_name == "attach_multiuser": + return await self._attach_multiuser(arguments) + elif tool_name == "add_user": + return await self._add_user(arguments) else: raise Exception(f"Unknown tool: {tool_name}") @@ -259,6 +321,51 @@ class ScreenServer(BaseMCPServer): else: return self.content_text(f"Failed to kill session: {result.stderr}") + async def _enable_multiuser(self, args: Dict[str, Any]) -> Dict[str, Any]: + """Enable multiuser mode for a screen session.""" + session = args["session"] + + # Enable multiuser mode + cmd = ["screen", "-S", session, "-X", "multiuser", "on"] + result = await self._run_command(cmd) + + if result.returncode == 0: + return self.content_text(f"Enabled multiuser mode for session '{session}'") + else: + return self.content_text(f"Failed to enable multiuser mode: {result.stderr}") + + async def _attach_multiuser(self, args: Dict[str, Any]) -> Dict[str, Any]: + """Provide instructions for attaching to a multiuser session.""" + session = args["session"] + user = args.get("user", "") + + # Check if session exists and is multiuser + check_result = await self._run_command(["screen", "-ls", session]) + if session not in check_result.stdout: + return self.content_text(f"Session '{session}' not found") + + # Provide attach command + if user: + attach_cmd = f"screen -x {user}/{session}" + else: + attach_cmd = f"screen -x {session}" + + return self.content_text(f"To attach to multiuser session '{session}', run: {attach_cmd}") + + async def _add_user(self, args: Dict[str, Any]) -> Dict[str, Any]: + """Add a user to a multiuser screen session.""" + session = args["session"] + user = args["user"] + + # Add user to session access control list + cmd = ["screen", "-S", session, "-X", "acladd", user] + result = await self._run_command(cmd) + + if result.returncode == 0: + return self.content_text(f"Added user '{user}' to session '{session}'") + else: + return self.content_text(f"Failed to add user: {result.stderr}") + async def _run_command(self, cmd: List[str]) -> subprocess.CompletedProcess: """Run a command and return the result.""" return await asyncio.to_thread( diff --git a/test_screen_multiuser.py b/test_screen_multiuser.py new file mode 100644 index 0000000..1c37e28 --- /dev/null +++ b/test_screen_multiuser.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +"""Test multiuser functionality in screen server.""" + +import asyncio +import json +import sys +import os + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from mcp_browser import MCPBrowser + + +async def test_screen_multiuser(): + """Test screen multiuser session functionality.""" + browser = MCPBrowser(enable_builtin_servers=True) + + await browser.initialize() + + print("=== Testing Screen Multiuser Functionality ===\n") + + # Create a test session + print("1. Creating test session...") + response = await browser.call({ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "create_session", + "arguments": { + "name": "multiuser-test", + "command": "bash" + } + } + }) + print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}") + + # Enable multiuser mode + print("\n2. Enabling multiuser mode...") + response = await browser.call({ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "enable_multiuser", + "arguments": { + "session": "multiuser-test" + } + } + }) + print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}") + + # Add a user to the session + print("\n3. Adding user to session...") + response = await browser.call({ + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": { + "name": "add_user", + "arguments": { + "session": "multiuser-test", + "user": "testuser" + } + } + }) + print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}") + + # Get attach instructions + print("\n4. Getting attach instructions...") + response = await browser.call({ + "jsonrpc": "2.0", + "id": 4, + "method": "tools/call", + "params": { + "name": "attach_multiuser", + "arguments": { + "session": "multiuser-test", + "user": "testuser" + } + } + }) + print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}") + + # Execute a command in the multiuser session + print("\n5. Executing command in multiuser session...") + response = await browser.call({ + "jsonrpc": "2.0", + "id": 5, + "method": "tools/call", + "params": { + "name": "execute", + "arguments": { + "session": "multiuser-test", + "command": "echo 'Multiuser session test - Hello World!'" + } + } + }) + print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}") + + # Wait a moment for output + await asyncio.sleep(0.5) + + # Test peek functionality with multiuser session + print("\n6. Testing peek with multiuser session...") + response = await browser.call({ + "jsonrpc": "2.0", + "id": 6, + "method": "tools/call", + "params": { + "name": "peek", + "arguments": { + "session": "multiuser-test", + "lines": 10 + } + } + }) + + if "result" in response: + output = response["result"]["content"][0]["text"] + print(f" Output:\n{output}") + + # Check if we can see the executed command output + if "Multiuser session test" in output: + print("\n✓ Peek works correctly with multiuser session!") + else: + print("\n⚠ Peek output might not show recent commands") + else: + print(f" Error: {response}") + + # List sessions to see multiuser status + print("\n7. Listing sessions...") + response = await browser.call({ + "jsonrpc": "2.0", + "id": 7, + "method": "tools/call", + "params": { + "name": "list_sessions", + "arguments": {} + } + }) + + if "result" in response: + sessions_output = response["result"]["content"][0]["text"] + print(f" Sessions:\n{sessions_output}") + + # Check if multiuser session is listed + if "multiuser-test" in sessions_output: + print("\n✓ Multiuser session is listed!") + else: + print("\n⚠ Session not found in list") + else: + print(f" Error: {response}") + + # Clean up + print("\n8. Cleaning up...") + response = await browser.call({ + "jsonrpc": "2.0", + "id": 8, + "method": "tools/call", + "params": { + "name": "kill_session", + "arguments": {"session": "multiuser-test"} + } + }) + print(f" Result: {response.get('result', {}).get('content', [{}])[0].get('text', 'Error')}") + + await browser.close() + + +if __name__ == "__main__": + asyncio.run(test_screen_multiuser()) \ No newline at end of file