MCP Terminal

Instructions

Implementation Reference

• src/mcp_terminal/tools/terminal.py:118-159 (handler)

  The MCP tool handler function 'execute_command' decorated with @mcp.tool(name="execute_command"). Performs command filtering using CommandFilter, executes via self.controller.execute_command, and maps result to ExecuteCommandResponse.
  @mcp.tool(name="execute_command", description="Executes a terminal command") async def execute_command( command: str, wait_for_output: bool = True, timeout: int = 10, ) -> ExecuteCommandResponse: try: # Check if command is allowed is_allowed, reason = self.command_filter.is_command_allowed(command) if not is_allowed: logger.warning( f"Command execution denied: {command}. Reason: {reason}" ) return ExecuteCommandResponse( success=False, error=f"Command not allowed: {reason}", ) # Ensure we have a controller if not self.controller: self._init_controller() # Execute the command result = await self.controller.execute_command( command, wait_for_output, timeout ) # Convert to response model return ExecuteCommandResponse( success=result.get("success", False), output=result.get("output"), error=result.get("error"), return_code=result.get("return_code"), warning=result.get("warning"), ) except Exception as e: logger.error(f"Error executing command: {e}") return ExecuteCommandResponse( success=False, error=f"Error executing command: {str(e)}" )
• src/mcp_terminal/tools/terminal.py:24-34 (schema)

  ExecuteCommandRequest Pydantic model defining the input parameters: command (str), wait_for_output (bool, default True), timeout (int, default 10).
  class ExecuteCommandRequest(BaseModel): """Request model for executing a terminal command.""" command: str = Field(..., description="The command to execute in the terminal") wait_for_output: bool = Field( True, description="Whether to wait for the command output" ) timeout: int = Field( 10, description="Timeout in seconds for waiting for the command output" )
• src/mcp_terminal/tools/terminal.py:36-50 (schema)

  ExecuteCommandResponse Pydantic model defining the output: success (bool), output (str|None), error (str|None), return_code (int|None), warning (str|None).
  class ExecuteCommandResponse(BaseModel): """Response model for the executed terminal command.""" success: bool = Field( ..., description="Whether the command execution was successful" ) output: Optional[str] = Field(None, description="The command output if available") error: Optional[str] = Field( None, description="Error message if the command failed" ) return_code: Optional[int] = Field( None, description="The command return code if available" ) warning: Optional[str] = Field(None, description="Warning message if any")
• src/mcp_terminal/server.py:93-107 (registration)

  In MCPTerminalServer.register_tools(): instantiates TerminalTool with config, then calls terminal_tool.register_mcp(self.mcp) which defines and registers the execute_command tool.
  # Create and register the terminal tool terminal_tool = TerminalTool( self.controller_type, whitelist_file=self.whitelist_file, blacklist_file=self.blacklist_file, whitelist_mode=self.whitelist_mode, ) file_tool = FileTool() terminal_tool.register_mcp(self.mcp) file_tool.register_mcp(self.mcp) self.tools["terminal"] = terminal_tool self.tools["file"] = file_tool self.tools_registered = True logger.info("Terminal tool registered with MCP server")
• src/mcp_terminal/controllers/subprocess.py:16-73 (helper)

  SubprocessTerminalController.execute_command: core implementation using asyncio.create_subprocess_shell to execute commands, capture stdout/stderr/returncode with timeout support. Default cross-platform controller.
  async def execute_command( self, command: str, wait_for_output: bool = True, timeout: int = 10 ) -> Dict[str, Any]: """ Execute a command using subprocess. Args: command: The command to execute wait_for_output: Whether to wait for output timeout: Timeout in seconds Returns: A dictionary with the result of the command execution """ try: # Create subprocess process = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) if wait_for_output: try: # Wait for the process to complete with timeout stdout, stderr = await asyncio.wait_for( process.communicate(), timeout=timeout ) return { "success": process.returncode == 0, "output": stdout.decode("utf-8", errors="replace"), "error": stderr.decode("utf-8", errors="replace"), "return_code": process.returncode, } except asyncio.TimeoutError: # Kill the process if it times out try: process.kill() except ProcessLookupError: pass return { "success": False, "error": f"Command timed out after {timeout} seconds", } else: # Don't wait for output return { "success": True, "output": "Command sent (output not captured)", } except Exception as e: return { "success": False, "error": f"Error executing command: {str(e)}", }