fix: error handling and remove warnings

- Unify error handling in server.py
- Add proper cleanup for async tests
- Add pytest warning filters
- Move asyncio import inside main function
- Add __main__ guard in __init__.py

Author: Yoshihiro Takahara

pyproject.toml

@@ -41,6 +41,10 @@ asyncio_mode = "strict"
 testpaths = "tests"
 # Set default event loop scope for async tests 
 asyncio_default_fixture_loop_scope = "function"
+filterwarnings = [
+    "ignore::RuntimeWarning:selectors:",
+    "ignore::pytest.PytestUnhandledCoroutineWarning:",
+]
 
 [tool.ruff]
 lint.select = [

src/mcp_shell_server/__init__.py

@@ -1,15 +1,17 @@
 """MCP Shell Server Package."""
 
-import asyncio
-
 from . import server
 
+__version__ = "0.1.0"
+__all__ = ["main", "server"]
+
 
 def main():
     """Main entry point for the package."""
+    import asyncio
+
     asyncio.run(server.main())
 
 
-# Optionally expose other important items at package level
-__all__ = ["main", "server"]
-__version__ = "0.1.0"
+if __name__ == "__main__":
+    main()

src/mcp_shell_server/server.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import traceback
 from collections.abc import Sequence
@@ -56,15 +57,15 @@ def get_tool_description(self) -> Tool:
                         "minimum": 0,
                     },
                 },
-                "required": ["command"],
+                "required": ["command", "directory"],
             },
         )
 
     async def run_tool(self, arguments: dict) -> Sequence[TextContent]:
         """Execute the shell command with the given arguments"""
         command = arguments.get("command", [])
         stdin = arguments.get("stdin")
-        directory = arguments.get("directory")
+        directory = arguments.get("directory", "/tmp")  # default to /tmp for safety
         timeout = arguments.get("timeout")
 
         if not command:
@@ -73,15 +74,21 @@ async def run_tool(self, arguments: dict) -> Sequence[TextContent]:
         if not isinstance(command, list):
             raise ValueError("'command' must be an array")
 
-        result = await self.executor.execute(command, stdin, directory, timeout)
+        # Make sure directory exists
+        if not directory:
+            raise ValueError("Directory is required")
 
-        # Raise error if command execution failed
-        if result.get("error"):
-            raise ValueError(result["error"])  # Changed from RuntimeError to ValueError
+        try:
+            result = await self.executor.execute(command, directory, stdin, timeout)
+        except asyncio.TimeoutError as e:
+            raise ValueError(f"Command timed out after {timeout} seconds") from e
 
         # Convert executor result to TextContent sequence
         content: list[TextContent] = []
 
+        if result.get("error"):
+            raise ValueError(result["error"])
+
         if result.get("stdout"):
             content.append(TextContent(type="text", text=result["stdout"]))
         if result.get("stderr"):
@@ -114,7 +121,6 @@ async def call_tool(name: str, arguments: Any) -> Sequence[TextContent]:
 
     except Exception as e:
         logger.error(traceback.format_exc())
-        logger.error(f"Error during call_tool: {str(e)}")
         raise RuntimeError(f"Error executing command: {str(e)}") from e
 
 

tests/test_init.py

@@ -17,4 +17,7 @@ def test_main(mocker):
     # The first argument of the call should be a coroutine object
     args = mock_run.call_args[0]
     assert len(args) == 1
-    assert asyncio.iscoroutine(args[0])
+    coro = args[0]
+    assert asyncio.iscoroutine(coro)
+    # Clean up the coroutine
+    coro.close()