fix: improve process manager and shell executor integration

- Fix command string formatting in shell executor
- Improve stderr handling
- Fix type hints and formatting in process manager
- Enhance pipeline execution
- All tests passing

Author: tumf

src/mcp_shell_server/process_manager.py

@@ -1,8 +1,9 @@
 """Process management for shell command execution."""
+
 import asyncio
 import logging
 import os
-from typing import Dict, IO, Optional, Tuple, Any, List
+from typing import IO, Any, Dict, List, Optional, Tuple
 
 
 class ProcessManager:
@@ -115,14 +116,18 @@ async def execute_pipeline(
 
                 try:
                     stdout, stderr = await self.execute_with_timeout(
-                        process, stdin=prev_stdout.decode() if prev_stdout else None, timeout=timeout
+                        process,
+                        stdin=prev_stdout.decode() if prev_stdout else None,
+                        timeout=timeout,
                     )
 
                     final_stderr += stderr if stderr else b""
                     if process.returncode != 0:
                         error_msg = stderr.decode("utf-8", errors="replace").strip()
                         if not error_msg:
-                            error_msg = f"Command failed with exit code {process.returncode}"
+                            error_msg = (
+                                f"Command failed with exit code {process.returncode}"
+                            )
                         raise ValueError(error_msg)
 
                     if i == len(commands) - 1:
@@ -140,12 +145,22 @@ async def execute_pipeline(
                     process.kill()
                     raise
 
-            return final_stdout, final_stderr, processes[-1].returncode if processes else 1
+            return (
+                final_stdout,
+                final_stderr,
+                (
+                    processes[-1].returncode
+                    if processes and processes[-1].returncode is not None
+                    else 1
+                ),
+            )
 
         finally:
             await self.cleanup_processes(processes)
 
-    async def cleanup_processes(self, processes: List[asyncio.subprocess.Process]) -> None:
+    async def cleanup_processes(
+        self, processes: List[asyncio.subprocess.Process]
+    ) -> None:
         """Clean up processes by killing them if they're still running.
 
         Args:

src/mcp_shell_server/shell_executor.py

@@ -12,6 +12,7 @@
 from mcp_shell_server.io_redirection_handler import IORedirectionHandler
 from mcp_shell_server.process_manager import ProcessManager
 
+
 class ShellExecutor:
     """
     Executes shell commands in a secure manner by validating against a whitelist.
@@ -226,10 +227,7 @@ async def execute(
             shell_cmd = f"{shell} -i -c {shlex.quote(shell_cmd)}"
 
             process = await self.process_manager.create_process(
-                shell_cmd,
-                directory,
-                stdout_handle=stdout_handle,
-                envs=envs
+                shell_cmd, directory, stdout_handle=stdout_handle, envs=envs
             )
 
             try:
@@ -248,9 +246,7 @@ async def communicate_with_timeout():
 
                 try:
                     stdout, stderr = await self.process_manager.execute_with_timeout(
-                        process,
-                        stdin=stdin,
-                        timeout=timeout
+                        process, stdin=stdin, timeout=timeout
                     )
 
                     # Close file handle if using file redirection
@@ -382,25 +378,27 @@ async def _execute_pipeline(
 
                 try:
                     # Execute the current command using ProcessManager
+                    shell_cmd = " ".join(map(shlex.quote, parsed_commands[i]))
                     process = await self.process_manager.create_process(
-                        cmd_str,
+                        shell_cmd,
                         directory,
                         stdout_handle=(
                             asyncio.subprocess.PIPE
                             if i < len(parsed_commands) - 1 or not last_stdout
                             else last_stdout
                         ),
-                        envs=envs
+                        envs=envs,
                     )
-                    
                     stdout, stderr = await self.process_manager.execute_with_timeout(
                         process,
                         stdin=prev_stdout.decode() if prev_stdout else None,
-                        timeout=timeout
+                        timeout=timeout,
                     )
 
                     # Collect stderr and check return code
-                    final_stderr += stderr if stderr else b""
+                    final_stderr = final_stderr + (
+                        stderr if stderr is not None else b""
+                    )
                     if process.returncode != 0:
                         error_msg = stderr.decode("utf-8", errors="replace").strip()
                         if not error_msg: